From 0f1d45dc36fe58aaa46b21b91748c00c058e3215 Mon Sep 17 00:00:00 2001 From: Rory Byrne Date: Mon, 3 Aug 2026 17:52:29 +0100 Subject: [PATCH 1/5] feat(ingest): ingester identity + versioned releases MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Domain layer for the record-provenance fix (#180, closes the record-level half of #164). Hooks have had traceable code identity since #145: every features.* row carries run_id -> hook_run -> hook_release -> (image, digest, config, source_ref). Records did not. The ingester's image and digest lived inline on the convention as a mutable blob overwritten on every redeploy, so record -> ingest_run -> ??? dead-ended, and "which code asserted this metadata, at what version" was unanswerable. Mirrors the hook registry rather than inventing a parallel shape: - `Ingester` aggregate — name, target schema, live-release pointer. Bound to the bare schema id, not a versioned SchemaId, so an ingester keeps producing records as its schema gains versions. - `IngesterRelease` — immutable, monotonic per name, reusing `OciConfig` for the runtime. "How a container runs" is one shape; both runners consume it identically. - `IngesterReleaseOutcome.created` — decided under the adapter's row lock, so 201-vs-200 needs no racy pre-check. - `IngesterRegistry` port + `IngesterRegistryService`. Also factors `PgName` out of `shared/model/names.py`: `HookName` and `FeatureName` had independently copied the same regex, validator and `__str__`, and `IngesterName` would have been a third copy. The 40-char cap and its reasoning now live in one place. Adapter, tables and wiring follow in the next commits on this branch. --- server/osa/domain/ingest/model/ingester.py | 35 ++++++ .../domain/ingest/model/ingester_release.py | 62 ++++++++++ .../domain/ingest/port/ingester_registry.py | 88 +++++++++++++ .../ingest/service/ingester_registry.py | 70 +++++++++++ server/osa/domain/shared/model/hook.py | 58 ++------- server/osa/domain/shared/model/names.py | 52 ++++++++ server/osa/domain/shared/model/source.py | 13 +- .../domain/ingest/test_ingester_registry.py | 117 ++++++++++++++++++ 8 files changed, 448 insertions(+), 47 deletions(-) create mode 100644 server/osa/domain/ingest/model/ingester.py create mode 100644 server/osa/domain/ingest/model/ingester_release.py create mode 100644 server/osa/domain/ingest/port/ingester_registry.py create mode 100644 server/osa/domain/ingest/service/ingester_registry.py create mode 100644 server/osa/domain/shared/model/names.py create mode 100644 server/tests/unit/domain/ingest/test_ingester_registry.py diff --git a/server/osa/domain/ingest/model/ingester.py b/server/osa/domain/ingest/model/ingester.py new file mode 100644 index 00000000..bd38f937 --- /dev/null +++ b/server/osa/domain/ingest/model/ingester.py @@ -0,0 +1,35 @@ +"""Ingester aggregate — stable identity, target schema, live pointer (#180). + +Mirrors :class:`~osa.domain.validation.model.hook.Hook`. Where a hook owns a +fixed *feature contract*, an ingester owns the *schema it produces records for*; +both carry a ``live_release_id`` naming the currently active release, advanced +under a row lock by the registry adapter. + +``schema_id`` is the bare schema id, not a versioned :class:`SchemaId`: an +ingester keeps producing records as its schema gains versions, so binding it to +one version would break on every field addition. +""" + +from __future__ import annotations + +from datetime import datetime + +from osa.domain.ingest.model.ingester_release import IngesterReleaseId +from osa.domain.shared.model.aggregate import Aggregate +from osa.domain.shared.model.source import IngesterName +from osa.domain.shared.model.srn import LocalId + + +class Ingester(Aggregate): + """Stable ingester identity + target schema + live-release pointer.""" + + model_config = {"frozen": True} + + name: IngesterName + schema_id: LocalId + live_release_id: IngesterReleaseId | None = None + created_at: datetime + + def with_live_release(self, release_id: IngesterReleaseId) -> "Ingester": + """Return a copy whose live pointer references *release_id*.""" + return self.model_copy(update={"live_release_id": release_id}) diff --git a/server/osa/domain/ingest/model/ingester_release.py b/server/osa/domain/ingest/model/ingester_release.py new file mode 100644 index 00000000..eb0d6a03 --- /dev/null +++ b/server/osa/domain/ingest/model/ingester_release.py @@ -0,0 +1,62 @@ +"""IngesterRelease — the immutable, versioned ingester artifact (#180). + +The exact mirror of :class:`~osa.domain.validation.model.hook_release.HookRelease`, +and for the same reason: a release captures *what code* ran, so a row it produced +can name the image, digest, config and build ref that produced it. + +Hooks have had this since #145 — every ``features.*`` row carries a ``run_id`` +that resolves to a release. Records did not: the ingester's image and digest +lived inline on the convention as a mutable blob, overwritten on every redeploy, +so "which code asserted this metadata, at what version" was unanswerable. That +asymmetry is the core of #164. This type closes it. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime +from typing import Annotated, NewType +from uuid import UUID + +from pydantic import Field + +from osa.domain.shared.model.entity import Entity +from osa.domain.shared.model.hook import OciConfig +from osa.domain.shared.model.source import IngesterName + +IngesterReleaseId = NewType("IngesterReleaseId", UUID) + + +class IngesterRelease(Entity): + """Immutable, versioned ingester artifact. + + ``runtime`` + ``source_ref`` are the deployable unit; ``version`` is + monotonic per ``ingester_name``. Reuses :class:`OciConfig` rather than + defining a parallel ingester runtime — "how a container runs" is one shape, + and the ingester and hook runners consume it identically. + """ + + model_config = {"frozen": True} + + id: IngesterReleaseId + ingester_name: IngesterName + version: int + runtime: Annotated[OciConfig, Field(discriminator="type")] + source_ref: str + built_by: str | None = None + built_at: datetime + + +@dataclass(frozen=True) +class IngesterReleaseOutcome: + """Result of minting a release. + + ``created`` is ``True`` when a new version was minted, ``False`` for an + idempotent no-op (the digest already existed). It is decided *inside* the + registry's row lock, so it is correct under concurrent identical + submissions — the caller maps it to HTTP 201 vs 200 without a racy + pre-check. + """ + + release: IngesterRelease + created: bool diff --git a/server/osa/domain/ingest/port/ingester_registry.py b/server/osa/domain/ingest/port/ingester_registry.py new file mode 100644 index 00000000..63223769 --- /dev/null +++ b/server/osa/domain/ingest/port/ingester_registry.py @@ -0,0 +1,88 @@ +"""Port for the ingester registry (#180). + +Persists ingester identities, their immutable versioned releases, and the live +pointer. The adapter owns the concurrency-critical bits — gap-free monotonic +version assignment and live-pointer advance under a row lock on the +``ingesters`` row — exactly as +:class:`~osa.domain.validation.port.hook_registry.HookRegistry` does. +""" + +from __future__ import annotations + +from abc import abstractmethod +from typing import Protocol + +from osa.domain.ingest.model.ingester import Ingester +from osa.domain.ingest.model.ingester_release import ( + IngesterRelease, + IngesterReleaseId, + IngesterReleaseOutcome, +) +from osa.domain.shared.model.hook import OciConfig +from osa.domain.shared.model.source import IngesterName +from osa.domain.shared.model.srn import LocalId +from osa.domain.shared.port import Port + + +class IngesterRegistry(Port, Protocol): + @abstractmethod + async def upsert_identity(self, name: IngesterName, schema_id: LocalId) -> Ingester: + """Create the ingester identity if absent; return the existing or new one. + + If the ingester exists bound to a **different** schema, raise + ``ConflictError``: records already published under this name belong to + the original schema, and silently repointing would orphan them. + """ + ... + + @abstractmethod + async def create_release( + self, + name: IngesterName, + runtime: OciConfig, + source_ref: str, + built_by: str | None, + ) -> IngesterReleaseOutcome: + """Mint the next release for an existing ingester and advance live. + + Idempotent on ``(name, digest)``: re-submitting an existing digest + returns the existing release without minting a version or moving the + pointer, with ``created == False``. Version assignment and pointer + advance happen under a row lock on the ``ingesters`` row so concurrent + submitters serialize, and ``created`` is decided under that same lock so + it is race-free. + """ + ... + + @abstractmethod + async def set_live(self, name: IngesterName, version: int) -> Ingester: + """Repoint live to an existing release of the ingester (rollback / pin).""" + ... + + @abstractmethod + async def get_ingester(self, name: IngesterName) -> Ingester | None: ... + + @abstractmethod + async def list_ingesters(self) -> list[Ingester]: ... + + @abstractmethod + async def list_releases(self, name: IngesterName) -> list[IngesterRelease]: + """All releases for an ingester, version-descending. Empty if absent.""" + ... + + @abstractmethod + async def get_release(self, name: IngesterName, version: int) -> IngesterRelease | None: ... + + @abstractmethod + async def get_release_by_id(self, release_id: IngesterReleaseId) -> IngesterRelease | None: + """Resolve a release by id — the record → run → release provenance hop.""" + ... + + @abstractmethod + async def resolve_live(self, name: IngesterName) -> IngesterRelease | None: + """The ingester's current live release, resolved once at run start. + + Snapshotted onto the run so a redeploy mid-run cannot change which code + the run is attributed to. ``None`` when the ingester has no release. + """ + ... diff --git a/server/osa/domain/ingest/service/ingester_registry.py b/server/osa/domain/ingest/service/ingester_registry.py new file mode 100644 index 00000000..a791cfb8 --- /dev/null +++ b/server/osa/domain/ingest/service/ingester_registry.py @@ -0,0 +1,70 @@ +"""IngesterRegistryService — business logic for the ingester registry (#180). + +Thin orchestration over the :class:`IngesterRegistry` port, mirroring +:class:`~osa.domain.validation.service.hook_registry.HookRegistryService`. The +concurrency-critical version and pointer mechanics live in the adapter. +""" + +from __future__ import annotations + +from osa.domain.ingest.model.ingester import Ingester +from osa.domain.ingest.model.ingester_release import ( + IngesterRelease, + IngesterReleaseId, + IngesterReleaseOutcome, +) +from osa.domain.ingest.port.ingester_registry import IngesterRegistry +from osa.domain.shared.error import NotFoundError +from osa.domain.shared.model.hook import OciConfig +from osa.domain.shared.model.source import IngesterName +from osa.domain.shared.model.srn import LocalId +from osa.domain.shared.service import Service + + +class IngesterRegistryService(Service): + registry: IngesterRegistry + + async def upsert_identity(self, name: IngesterName, schema_id: LocalId) -> Ingester: + """Create the ingester identity if absent; reject a differing schema.""" + return await self.registry.upsert_identity(name, schema_id) + + async def create_release( + self, + name: IngesterName, + runtime: OciConfig, + source_ref: str, + built_by: str | None = None, + ) -> IngesterReleaseOutcome: + """Mint vN+1 for an existing ingester (idempotent on digest); advance live.""" + return await self.registry.create_release(name, runtime, source_ref, built_by) + + async def set_live(self, name: IngesterName, version: int) -> Ingester: + """Repoint the live pointer to a prior release (rollback / pin).""" + return await self.registry.set_live(name, version) + + async def get_ingester(self, name: IngesterName) -> Ingester | None: + return await self.registry.get_ingester(name) + + async def list_ingesters(self) -> list[Ingester]: + return await self.registry.list_ingesters() + + async def list_releases(self, name: IngesterName) -> list[IngesterRelease]: + return await self.registry.list_releases(name) + + async def get_release(self, name: IngesterName, version: int) -> IngesterRelease | None: + return await self.registry.get_release(name, version) + + async def get_release_by_id(self, release_id: IngesterReleaseId) -> IngesterRelease | None: + return await self.registry.get_release_by_id(release_id) + + async def require_live_release(self, name: IngesterName) -> IngesterRelease: + """Resolve the live release for a run, or fail with a nameable reason. + + Callers start a run with this: an ingester that exists but has never had + a release is a deploy-time mistake, and a run attributed to no code at + all is exactly the provenance hole this registry closes. + """ + release = await self.registry.resolve_live(name) + if release is None: + raise NotFoundError(f"Ingester {name.root!r} has no live release") + return release diff --git a/server/osa/domain/shared/model/hook.py b/server/osa/domain/shared/model/hook.py index 59920d95..f9c99221 100644 --- a/server/osa/domain/shared/model/hook.py +++ b/server/osa/domain/shared/model/hook.py @@ -8,8 +8,9 @@ import re from typing import Annotated, Any, ClassVar, Literal -from pydantic import ConfigDict, Field, RootModel, field_validator +from pydantic import Field +from osa.domain.shared.model.names import PgName from osa.domain.shared.model.value import ValueObject # Lowercase alphanumeric + underscore, starting with a letter, max 63 chars. @@ -17,65 +18,30 @@ PgIdentifier = Annotated[str, Field(pattern=r"^[a-z][a-z0-9_]{0,62}$")] -class HookName(RootModel[str]): - """A hook's stable name — a frozen ``RootModel`` (#145). +class HookName(PgName): + """A hook's stable name (#145). - Promoted from a bare ``Annotated[str, …]`` to a nominal type so the type - checker can distinguish a hook name from any other string and the regex is - enforced at construction. Frozen, so it is hashable and usable as a dict key - (``dict[HookName, …]``). Use ``.root`` where a plain ``str`` is required - (PG identifiers built without interpolation, dict keys handed to infra). - - Hook names compose into PG identifiers alongside fixed prefixes/suffixes — - notably the per-hook FK constraint ``fk_features_{name}_record_srn`` (23 - chars of overhead). PG's identifier limit is 63 chars, so cap hook names at - 40 to keep every derived identifier inside the limit without surprise - truncation. Column names use plain :data:`PgIdentifier` because they don't - get composed into longer names. + A nominal type so the type checker can distinguish a hook name from any + other string and the regex is enforced at construction. Column names use + plain :data:`PgIdentifier` instead, because they don't get composed into + longer identifiers. """ - model_config = ConfigDict(frozen=True) - - _re: ClassVar[re.Pattern] = re.compile(r"^[a-z][a-z0-9_]{0,39}$") - - @field_validator("root") - @classmethod - def _validate(cls, v: str) -> str: - if not cls._re.match(v): - raise ValueError("invalid hook name: 1–40 chars of [a-z0-9_], starting with a letter") - return v - - def __str__(self) -> str: - return self.root + kind: ClassVar[str] = "hook name" -class FeatureName(RootModel[str]): +class FeatureName(PgName): """Identity of a feature table on the read surface (#145). A hook produces exactly one feature table, addressed at ``/data/{schema}/{feature}``; its name equals the producing hook's name but is a **distinct nominal type** so the read/feature side never traffics in - "hook". Same PG-identifier rules as :class:`HookName` (≤40 chars). Convert a - producing hook's name once at the hook→feature boundary + "hook". Convert a producing hook's name once at the hook→feature boundary (``FeatureName(hook.name.root)``); reserved-slot names are already rejected upstream on the hook identity. """ - model_config = ConfigDict(frozen=True) - - _re: ClassVar[re.Pattern] = re.compile(r"^[a-z][a-z0-9_]{0,39}$") - - @field_validator("root") - @classmethod - def _validate(cls, v: str) -> str: - if not cls._re.match(v): - raise ValueError( - "invalid feature name: 1–40 chars of [a-z0-9_], starting with a letter" - ) - return v - - def __str__(self) -> str: - return self.root + kind: ClassVar[str] = "feature name" _MEMORY_RE = re.compile(r"^(\d+(?:\.\d+)?)(g|m|k)?i?$") diff --git a/server/osa/domain/shared/model/names.py b/server/osa/domain/shared/model/names.py new file mode 100644 index 00000000..f45ced7c --- /dev/null +++ b/server/osa/domain/shared/model/names.py @@ -0,0 +1,52 @@ +"""Nominal name types for things that compose into PostgreSQL identifiers. + +Hooks, features and ingesters all name the same *kind* of thing — a stable, +lowercase identifier that ends up inside a PG object name — and all three had +independently copied the same regex, validator and ``__str__``. :class:`PgName` +holds that shape once; each subclass exists purely so the type checker can tell +a hook name from a feature name from an ingester name. + +The 40-character cap is not arbitrary. These names compose into PG identifiers +alongside fixed prefixes and suffixes — the widest being the per-hook FK +constraint ``fk_features_{name}_record_srn`` at 23 characters of overhead. +PG's identifier limit is 63, so capping names at 40 keeps every derived +identifier inside the limit without surprise truncation. +""" + +from __future__ import annotations + +import re +from typing import ClassVar + +from pydantic import ConfigDict, RootModel, field_validator + +MAX_NAME_LENGTH = 40 + +_NAME_PATTERN = re.compile(rf"^[a-z][a-z0-9_]{{0,{MAX_NAME_LENGTH - 1}}}$") + + +class PgName(RootModel[str]): + """A stable name that is safe to compose into a PG identifier. + + Frozen, so instances are hashable and usable as dict keys + (``dict[HookName, ...]``). Use ``.root`` where a plain ``str`` is required — + PG identifiers built without interpolation, dict keys handed to infra. + """ + + model_config = ConfigDict(frozen=True) + + #: Names the kind of thing in validation errors ("hook name", "feature name"). + kind: ClassVar[str] = "name" + + @field_validator("root") + @classmethod + def _validate(cls, value: str) -> str: + if not _NAME_PATTERN.match(value): + raise ValueError( + f"invalid {cls.kind}: 1–{MAX_NAME_LENGTH} chars of [a-z0-9_], " + "starting with a letter" + ) + return value + + def __str__(self) -> str: + return self.root diff --git a/server/osa/domain/shared/model/source.py b/server/osa/domain/shared/model/source.py index 49f73e59..3192eb62 100644 --- a/server/osa/domain/shared/model/source.py +++ b/server/osa/domain/shared/model/source.py @@ -1,12 +1,23 @@ """Shared source domain models used across deposition and ingest domains.""" -from typing import Annotated, Any, Literal, Union +from typing import Annotated, Any, ClassVar, Literal, Union from pydantic import Discriminator, Field, Tag, field_validator +from osa.domain.shared.model.names import PgName from osa.domain.shared.model.value import ValueObject +class IngesterName(PgName): + """An ingester's stable name (#180). + + Lives in the shared kernel because the deploy path (deposition domain) + registers ingesters that the ingest domain later runs. + """ + + kind: ClassVar[str] = "ingester name" + + class IngesterLimits(ValueObject): """Resource limits for ingester container execution.""" diff --git a/server/tests/unit/domain/ingest/test_ingester_registry.py b/server/tests/unit/domain/ingest/test_ingester_registry.py new file mode 100644 index 00000000..6e73c135 --- /dev/null +++ b/server/tests/unit/domain/ingest/test_ingester_registry.py @@ -0,0 +1,117 @@ +"""Ingester identity + versioned releases — the record-provenance anchor (#180). + +Records asserted by an ingester had no traceable code identity: the image and +digest lived inline on the convention as a mutable blob, overwritten on every +redeploy, so ``record → ingest_run → ???`` dead-ended. Hooks already had +``feature row → hook_run → hook_release → (image, digest, config, source_ref)``. + +These tests pin the ingester half of that symmetry. +""" + +from datetime import UTC, datetime +from uuid import UUID, uuid4 + +import pytest + +from osa.domain.ingest.model.ingester import Ingester +from osa.domain.ingest.model.ingester_release import ( + IngesterRelease, + IngesterReleaseId, + IngesterReleaseOutcome, +) +from osa.domain.shared.model.hook import OciConfig, OciLimits +from osa.domain.shared.model.names import MAX_NAME_LENGTH +from osa.domain.shared.model.source import IngesterName +from osa.domain.shared.model.srn import LocalId + + +def _runtime(digest: str = "sha256:abc") -> OciConfig: + return OciConfig( + image="ghcr.io/example/ncbi-ingester:v1", + digest=digest, + config={}, + limits=OciLimits(timeout_seconds=3600, memory="2g", cpu="1.0"), + ) + + +def _release(version: int = 1, digest: str = "sha256:abc") -> IngesterRelease: + return IngesterRelease( + id=IngesterReleaseId(uuid4()), + ingester_name=IngesterName("from_ncbi"), + version=version, + runtime=_runtime(digest), + source_ref="git:deadbee", + built_by="ci", + built_at=datetime.now(UTC), + ) + + +def _ingester(live_release_id: IngesterReleaseId | None = None) -> Ingester: + return Ingester( + name=IngesterName("from_ncbi"), + schema_id=LocalId("strain"), + live_release_id=live_release_id, + created_at=datetime.now(UTC), + ) + + +class TestIngesterName: + def test_accepts_a_pg_safe_name(self): + assert IngesterName("from_ncbi").root == "from_ncbi" + + @pytest.mark.parametrize( + "invalid", + ["From_NCBI", "9lives", "has-dash", "", "a" * (MAX_NAME_LENGTH + 1)], + ) + def test_rejects_names_that_are_unsafe_as_identifiers(self, invalid: str): + with pytest.raises(ValueError, match="ingester name"): + IngesterName(invalid) + + def test_is_hashable_so_it_can_key_a_dict(self): + assert {IngesterName("from_ncbi"): 1}[IngesterName("from_ncbi")] == 1 + + +class TestIngesterIdentity: + def test_declares_the_schema_it_produces_records_for(self): + assert _ingester().schema_id == LocalId("strain") + + def test_starts_with_no_live_release(self): + assert _ingester().live_release_id is None + + def test_with_live_release_returns_a_repointed_copy(self): + ingester = _ingester() + release_id = IngesterReleaseId(uuid4()) + + repointed = ingester.with_live_release(release_id) + + assert repointed.live_release_id == release_id + assert ingester.live_release_id is None, "the original must not be mutated" + + def test_is_frozen(self): + with pytest.raises(ValueError): + _ingester().name = IngesterName("other") + + +class TestIngesterRelease: + def test_carries_the_full_code_identity_that_produced_a_record(self): + release = _release() + + assert release.runtime.image == "ghcr.io/example/ncbi-ingester:v1" + assert release.runtime.digest == "sha256:abc" + assert release.source_ref == "git:deadbee" + + def test_is_frozen_so_a_published_release_cannot_be_rewritten(self): + with pytest.raises(ValueError): + _release().version = 2 + + def test_release_id_is_a_uuid(self): + assert isinstance(_release().id, UUID) + + +class TestIngesterReleaseOutcome: + def test_created_distinguishes_a_new_version_from_an_idempotent_no_op(self): + minted = IngesterReleaseOutcome(release=_release(), created=True) + no_op = IngesterReleaseOutcome(release=_release(), created=False) + + assert minted.created is True + assert no_op.created is False From 99d98fe79543d86ef051fca3a8551d6f8827061b Mon Sep 17 00:00:00 2001 From: Rory Byrne Date: Mon, 3 Aug 2026 18:00:55 +0100 Subject: [PATCH 2/5] feat(ingest): persist ingester identities and releases MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Postgres adapter + DDL for the ingester registry, mirroring the hook registry's concurrency mechanics rather than inventing new ones: - Version assignment and live-pointer advance run under SELECT ... FOR UPDATE on the ingesters row, so concurrent release submissions for one ingester produce gap-free monotonic versions and never lose a pointer update. - upsert_identity is race-safe via ON CONFLICT DO NOTHING, then reads the winning row back and enforces the schema binding — repointing an ingester at a different schema is a ConflictError, because the records it already published belong to the original one. - Idempotency on (ingester_name, digest) is decided inside the row lock, so created is race-free. DDL is folded into the initial migration rather than added as an alter-chain (pre-launch, no data to migrate). alembic check reports no drift against a freshly migrated database. Uses NotFoundError/InfrastructureError for the post-insert reads rather than the hook adapter's bare assert. --- .../versions/c6d9f4c0c3ab_initial_schema.py | 59 ++++- .../repository/ingester_registry.py | 245 ++++++++++++++++++ .../osa/infrastructure/persistence/tables.py | 54 ++++ 3 files changed, 357 insertions(+), 1 deletion(-) create mode 100644 server/osa/infrastructure/persistence/repository/ingester_registry.py diff --git a/server/migrations/versions/c6d9f4c0c3ab_initial_schema.py b/server/migrations/versions/c6d9f4c0c3ab_initial_schema.py index cefa8bcc..10be398e 100644 --- a/server/migrations/versions/c6d9f4c0c3ab_initial_schema.py +++ b/server/migrations/versions/c6d9f4c0c3ab_initial_schema.py @@ -317,6 +317,58 @@ def upgrade() -> None: deferrable=True, initially="DEFERRED", ) + op.create_table( + "ingesters", + sa.Column("name", sa.String(length=40), nullable=False), + sa.Column("schema_id", sa.String(), nullable=False), + sa.Column("live_release_id", sa.UUID(), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.PrimaryKeyConstraint("name"), + ) + op.create_index("idx_ingesters_schema_id", "ingesters", ["schema_id"], unique=False) + op.create_table( + "ingester_releases", + sa.Column("id", sa.UUID(), nullable=False), + sa.Column("ingester_name", sa.String(length=40), nullable=False), + sa.Column("version", sa.Integer(), nullable=False), + sa.Column("image", sa.Text(), nullable=False), + sa.Column("digest", sa.Text(), nullable=False), + sa.Column( + "config", + postgresql.JSONB(astext_type=sa.Text()), + server_default=sa.text("'{}'"), + nullable=False, + ), + sa.Column("limits", postgresql.JSONB(astext_type=sa.Text()), nullable=False), + sa.Column("source_ref", sa.Text(), nullable=False), + sa.Column("built_by", sa.Text(), nullable=True), + sa.Column("built_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint( + ["ingester_name"], + ["ingesters.name"], + ), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("ingester_name", "digest", name="uq_ingester_releases_ingester_digest"), + sa.UniqueConstraint( + "ingester_name", "version", name="uq_ingester_releases_ingester_version" + ), + ) + op.create_index( + "idx_ingester_releases_ingester_version", + "ingester_releases", + ["ingester_name", sa.literal_column("version DESC")], + unique=False, + ) + # Same circular-dependency break as hooks above. + op.create_foreign_key( + "fk_ingesters_live_release_id", + "ingesters", + "ingester_releases", + ["live_release_id"], + ["id"], + deferrable=True, + initially="DEFERRED", + ) op.create_table( "identities", sa.Column("id", sa.String(), nullable=False), @@ -401,8 +453,13 @@ def upgrade() -> None: def downgrade() -> None: """Downgrade schema.""" # ### commands auto generated by Alembic - please adjust! ### - # Drop the live-pointer FK first so hook_releases can be dropped below. + # Drop the live-pointer FKs first so the release tables can be dropped below. op.drop_constraint("fk_hooks_live_release_id", "hooks", type_="foreignkey") + op.drop_constraint("fk_ingesters_live_release_id", "ingesters", type_="foreignkey") + op.drop_index("idx_ingester_releases_ingester_version", table_name="ingester_releases") + op.drop_table("ingester_releases") + op.drop_index("idx_ingesters_schema_id", table_name="ingesters") + op.drop_table("ingesters") op.drop_index("idx_hook_runs_release", table_name="hook_runs") op.drop_table("hook_runs") op.drop_index("ix_role_assignments_user_id", table_name="role_assignments") diff --git a/server/osa/infrastructure/persistence/repository/ingester_registry.py b/server/osa/infrastructure/persistence/repository/ingester_registry.py new file mode 100644 index 00000000..e421c1aa --- /dev/null +++ b/server/osa/infrastructure/persistence/repository/ingester_registry.py @@ -0,0 +1,245 @@ +"""Postgres adapter for the ingester registry (#180). + +Concurrency-critical operations — version assignment and live-pointer advance — +run under a ``SELECT ... FOR UPDATE`` row lock on the ``ingesters`` row, so +concurrent release submissions for one ingester produce gap-free monotonic +versions and never lose a pointer update. Identical reasoning, and identical +mechanics, to :class:`~osa.infrastructure.persistence.repository.hook_registry.PostgresHookRegistry`. +""" + +from __future__ import annotations + +from datetime import UTC, datetime +from typing import Any +from uuid import uuid4 + +from sqlalchemy import and_, func, insert, select, update +from sqlalchemy.dialects.postgresql import insert as pg_insert +from sqlalchemy.ext.asyncio import AsyncSession + +from osa.domain.ingest.model.ingester import Ingester +from osa.domain.ingest.model.ingester_release import ( + IngesterRelease, + IngesterReleaseId, + IngesterReleaseOutcome, +) +from osa.domain.ingest.port.ingester_registry import IngesterRegistry +from osa.domain.shared.error import ConflictError, InfrastructureError, NotFoundError +from osa.domain.shared.model.hook import OciConfig, OciLimits +from osa.domain.shared.model.source import IngesterName +from osa.domain.shared.model.srn import LocalId +from osa.infrastructure.persistence.tables import ( + ingester_releases_table, + ingesters_table, +) + + +class PostgresIngesterRegistry(IngesterRegistry): + def __init__(self, session: AsyncSession) -> None: + self.session = session + + @staticmethod + def _to_ingester(row: dict[str, Any]) -> Ingester: + live_release_id = row["live_release_id"] + return Ingester( + name=IngesterName(row["name"]), + schema_id=LocalId(row["schema_id"]), + live_release_id=( + IngesterReleaseId(live_release_id) if live_release_id is not None else None + ), + created_at=row["created_at"], + ) + + @staticmethod + def _to_release(row: dict[str, Any]) -> IngesterRelease: + return IngesterRelease( + id=IngesterReleaseId(row["id"]), + ingester_name=IngesterName(row["ingester_name"]), + version=row["version"], + runtime=OciConfig( + image=row["image"], + digest=row["digest"], + config=row["config"] or {}, + limits=OciLimits.model_validate(row["limits"]), + ), + source_ref=row["source_ref"], + built_by=row["built_by"], + built_at=row["built_at"], + ) + + async def _get_ingester_row(self, name: IngesterName) -> dict[str, Any] | None: + result = await self.session.execute( + select(ingesters_table).where(ingesters_table.c.name == name.root) + ) + row = result.mappings().first() + return dict(row) if row is not None else None + + async def upsert_identity(self, name: IngesterName, schema_id: LocalId) -> Ingester: + # Race-safe create-if-absent: two concurrent identical deploys must not + # collide on the primary key. ON CONFLICT DO NOTHING serializes on PG's + # internal lock — blocking on a concurrent uncommitted insert until it + # resolves — so the loser no-ops instead of erroring. The winning row is + # then read back and its schema binding enforced. + await self.session.execute( + pg_insert(ingesters_table) + .values( + name=name.root, + schema_id=schema_id.root, + live_release_id=None, + created_at=datetime.now(UTC), + ) + .on_conflict_do_nothing(index_elements=["name"]) + ) + await self.session.flush() + + row = await self._get_ingester_row(name) + if row is None: # pragma: no cover — the upsert above guarantees the row + raise InfrastructureError(f"ingester {name.root!r} missing immediately after upsert") + if row["schema_id"] != schema_id.root: + raise ConflictError( + f"Ingester {name.root!r} already produces records for schema " + f"{row['schema_id']!r}; repointing it to {schema_id.root!r} would orphan " + "the records it has already published", + code="ingester_schema_conflict", + ) + return self._to_ingester(row) + + async def create_release( + self, + name: IngesterName, + runtime: OciConfig, + source_ref: str, + built_by: str | None, + ) -> IngesterReleaseOutcome: + # Row-lock the ingester so concurrent releases serialize. Also asserts + # the ingester exists. + locked = await self.session.execute( + select(ingesters_table).where(ingesters_table.c.name == name.root).with_for_update() + ) + if locked.mappings().first() is None: + raise NotFoundError(f"Ingester not found: {name.root}") + + # Idempotency on (ingester_name, digest): return the existing release, no + # new version, pointer unchanged. Decided under the row lock, so `created` + # is race-free under concurrent identical submissions. + duplicate = await self.session.execute( + select(ingester_releases_table).where( + and_( + ingester_releases_table.c.ingester_name == name.root, + ingester_releases_table.c.digest == runtime.digest, + ) + ) + ) + duplicate_row = duplicate.mappings().first() + if duplicate_row is not None: + return IngesterReleaseOutcome( + release=self._to_release(dict(duplicate_row)), created=False + ) + + highest_version = await self.session.scalar( + select(func.coalesce(func.max(ingester_releases_table.c.version), 0)).where( + ingester_releases_table.c.ingester_name == name.root + ) + ) + next_version = int(highest_version or 0) + 1 + release_id = uuid4() + + await self.session.execute( + insert(ingester_releases_table).values( + id=release_id, + ingester_name=name.root, + version=next_version, + image=runtime.image, + digest=runtime.digest, + config=runtime.config, + limits=runtime.limits.model_dump(), + source_ref=source_ref, + built_by=built_by, + built_at=datetime.now(UTC), + ) + ) + await self.session.execute( + update(ingesters_table) + .where(ingesters_table.c.name == name.root) + .values(live_release_id=release_id) + ) + await self.session.flush() + + minted = await self.get_release(name, next_version) + if minted is None: # pragma: no cover — just inserted in this transaction + raise InfrastructureError( + f"ingester release {name.root}@v{next_version} missing immediately after insert" + ) + return IngesterReleaseOutcome(release=minted, created=True) + + async def set_live(self, name: IngesterName, version: int) -> Ingester: + locked = await self.session.execute( + select(ingesters_table).where(ingesters_table.c.name == name.root).with_for_update() + ) + if locked.mappings().first() is None: + raise NotFoundError(f"Ingester not found: {name.root}") + + target = await self.get_release(name, version) + if target is None: + raise NotFoundError(f"Release not found: {name.root}@v{version}") + + await self.session.execute( + update(ingesters_table) + .where(ingesters_table.c.name == name.root) + .values(live_release_id=target.id) + ) + await self.session.flush() + + row = await self._get_ingester_row(name) + if row is None: # pragma: no cover — locked above + raise InfrastructureError(f"ingester {name.root!r} vanished during set_live") + return self._to_ingester(row) + + async def get_ingester(self, name: IngesterName) -> Ingester | None: + row = await self._get_ingester_row(name) + return self._to_ingester(row) if row is not None else None + + async def list_ingesters(self) -> list[Ingester]: + result = await self.session.execute( + select(ingesters_table).order_by(ingesters_table.c.name) + ) + return [self._to_ingester(dict(row)) for row in result.mappings().all()] + + async def list_releases(self, name: IngesterName) -> list[IngesterRelease]: + result = await self.session.execute( + select(ingester_releases_table) + .where(ingester_releases_table.c.ingester_name == name.root) + .order_by(ingester_releases_table.c.version.desc()) + ) + return [self._to_release(dict(row)) for row in result.mappings().all()] + + async def get_release(self, name: IngesterName, version: int) -> IngesterRelease | None: + result = await self.session.execute( + select(ingester_releases_table).where( + and_( + ingester_releases_table.c.ingester_name == name.root, + ingester_releases_table.c.version == version, + ) + ) + ) + row = result.mappings().first() + return self._to_release(dict(row)) if row is not None else None + + async def get_release_by_id(self, release_id: IngesterReleaseId) -> IngesterRelease | None: + result = await self.session.execute( + select(ingester_releases_table).where(ingester_releases_table.c.id == release_id) + ) + row = result.mappings().first() + return self._to_release(dict(row)) if row is not None else None + + async def resolve_live(self, name: IngesterName) -> IngesterRelease | None: + result = await self.session.execute( + select(ingester_releases_table) + .join( + ingesters_table, + ingesters_table.c.live_release_id == ingester_releases_table.c.id, + ) + .where(ingesters_table.c.name == name.root) + ) + row = result.mappings().first() + return self._to_release(dict(row)) if row is not None else None diff --git a/server/osa/infrastructure/persistence/tables.py b/server/osa/infrastructure/persistence/tables.py index 102cdffa..9498f4c4 100644 --- a/server/osa/infrastructure/persistence/tables.py +++ b/server/osa/infrastructure/persistence/tables.py @@ -473,6 +473,60 @@ ) +# Ingester identity: which schema it produces records for, and which release is +# live (#180). The exact mirror of `hooks`/`hook_releases` — an ingester asserts +# records the way a hook asserts feature rows, and both need to name the code +# that did it. +ingesters_table = Table( + "ingesters", + metadata, + Column("name", String(40), primary_key=True), # IngesterName, globally unique + # Bare schema id, not `@`: an ingester keeps producing records as + # its schema gains versions. + Column("schema_id", String, nullable=False), + Column( + "live_release_id", + PGUUID(as_uuid=True), + ForeignKey( + "ingester_releases.id", + name="fk_ingesters_live_release_id", + use_alter=True, + deferrable=True, + initially="DEFERRED", + ), + nullable=True, + ), + Column("created_at", DateTime(timezone=True), nullable=False), +) + +Index("idx_ingesters_schema_id", ingesters_table.c.schema_id) + + +# Immutable, integer-versioned ingester artifact: what image runs, built from where. +ingester_releases_table = Table( + "ingester_releases", + metadata, + Column("id", PGUUID(as_uuid=True), primary_key=True), # IngesterReleaseId + Column("ingester_name", String(40), ForeignKey("ingesters.name"), nullable=False), + Column("version", Integer, nullable=False), # monotonic per ingester, gap-free + Column("image", Text, nullable=False), + Column("digest", Text, nullable=False), + Column("config", JSONB, nullable=False, server_default=text("'{}'")), + Column("limits", JSONB, nullable=False), + Column("source_ref", Text, nullable=False), # git SHA / build id (reproducibility) + Column("built_by", Text, nullable=True), + Column("built_at", DateTime(timezone=True), nullable=False), + UniqueConstraint("ingester_name", "version", name="uq_ingester_releases_ingester_version"), + UniqueConstraint("ingester_name", "digest", name="uq_ingester_releases_ingester_digest"), +) + +Index( + "idx_ingester_releases_ingester_version", + ingester_releases_table.c.ingester_name, + ingester_releases_table.c.version.desc(), +) + + # Append-only PURE execution record + per-row provenance anchor (design-revisions # §6). No execution-context columns: a feature row reaches its data origin via the # other arm of the join (record_srn → records.source); this is only "what code ran, From f25e17561bb5666df7cfa0a70a8ead34ef5ddcf4 Mon Sep 17 00:00:00 2001 From: Rory Byrne Date: Fri, 14 Aug 2026 19:54:27 +0100 Subject: [PATCH 3/5] feat(ingest): mint ingester releases on deploy, snapshot on run start MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wire the ingester registry into its callers, greenfield (#180 §1): - IngesterDefinition.name and .source_ref are required — a declared ingester IS an identity and promises provenance; the nameless and unversioned forms are unrepresentable, so no back-compat guards exist. - Deploy upserts the ingester identity and mints a release (idempotent on digest, advancing the live pointer), mirroring hooks. - start_ingest resolves the live release once and snapshots it on the run; ingest_runs.release_id is NOT NULL with a real FK. No live release ⇒ the run is refused. - Wire names (class names, slugs) normalise deterministically into registry identities via PgName.from_raw at the DTO boundary. - Migration e1a7d20c4f9b is an incremental revision (live archives upgrade in place; new policy — no more folding into initial_schema). It clears ingest_runs: pre-registry runs have no release to point at, and runs are re-runnable operational state, not archival data. --- .../versions/c6d9f4c0c3ab_initial_schema.py | 59 +--------- .../e1a7d20c4f9b_add_ingester_registry.py | 108 ++++++++++++++++++ .../deposition/command/create_convention.py | 7 +- .../domain/deposition/service/convention.py | 23 ++++ .../osa/domain/deposition/util/di/provider.py | 3 + server/osa/domain/ingest/model/ingest_run.py | 5 + server/osa/domain/ingest/service/ingest.py | 8 ++ server/osa/domain/shared/model/names.py | 19 ++- server/osa/domain/shared/model/source.py | 14 ++- server/osa/infrastructure/ingest/di.py | 7 ++ server/osa/infrastructure/persistence/di.py | 10 ++ .../persistence/repository/ingest.py | 3 + .../osa/infrastructure/persistence/tables.py | 8 ++ .../persistence/test_convention_repo.py | 3 + .../persistence/test_ingest_repo.py | 106 +++++++++++++---- .../test_bulk_publish_dual_write.py | 5 + .../workflow/test_process_batch.py | 11 +- .../deposition/test_convention_service.py | 1 + .../deposition/test_convention_service_v2.py | 68 ++++++++++- .../deposition/test_deploy_convention_dto.py | 34 +++++- .../unit/domain/ingest/test_get_ingestion.py | 4 + .../unit/domain/ingest/test_ingest_run.py | 3 + .../unit/domain/ingest/test_ingest_service.py | 56 ++++++++- .../k8s/test_k8s_ingester_runner.py | 4 +- 24 files changed, 470 insertions(+), 99 deletions(-) create mode 100644 server/migrations/versions/e1a7d20c4f9b_add_ingester_registry.py diff --git a/server/migrations/versions/c6d9f4c0c3ab_initial_schema.py b/server/migrations/versions/c6d9f4c0c3ab_initial_schema.py index 10be398e..cefa8bcc 100644 --- a/server/migrations/versions/c6d9f4c0c3ab_initial_schema.py +++ b/server/migrations/versions/c6d9f4c0c3ab_initial_schema.py @@ -317,58 +317,6 @@ def upgrade() -> None: deferrable=True, initially="DEFERRED", ) - op.create_table( - "ingesters", - sa.Column("name", sa.String(length=40), nullable=False), - sa.Column("schema_id", sa.String(), nullable=False), - sa.Column("live_release_id", sa.UUID(), nullable=True), - sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), - sa.PrimaryKeyConstraint("name"), - ) - op.create_index("idx_ingesters_schema_id", "ingesters", ["schema_id"], unique=False) - op.create_table( - "ingester_releases", - sa.Column("id", sa.UUID(), nullable=False), - sa.Column("ingester_name", sa.String(length=40), nullable=False), - sa.Column("version", sa.Integer(), nullable=False), - sa.Column("image", sa.Text(), nullable=False), - sa.Column("digest", sa.Text(), nullable=False), - sa.Column( - "config", - postgresql.JSONB(astext_type=sa.Text()), - server_default=sa.text("'{}'"), - nullable=False, - ), - sa.Column("limits", postgresql.JSONB(astext_type=sa.Text()), nullable=False), - sa.Column("source_ref", sa.Text(), nullable=False), - sa.Column("built_by", sa.Text(), nullable=True), - sa.Column("built_at", sa.DateTime(timezone=True), nullable=False), - sa.ForeignKeyConstraint( - ["ingester_name"], - ["ingesters.name"], - ), - sa.PrimaryKeyConstraint("id"), - sa.UniqueConstraint("ingester_name", "digest", name="uq_ingester_releases_ingester_digest"), - sa.UniqueConstraint( - "ingester_name", "version", name="uq_ingester_releases_ingester_version" - ), - ) - op.create_index( - "idx_ingester_releases_ingester_version", - "ingester_releases", - ["ingester_name", sa.literal_column("version DESC")], - unique=False, - ) - # Same circular-dependency break as hooks above. - op.create_foreign_key( - "fk_ingesters_live_release_id", - "ingesters", - "ingester_releases", - ["live_release_id"], - ["id"], - deferrable=True, - initially="DEFERRED", - ) op.create_table( "identities", sa.Column("id", sa.String(), nullable=False), @@ -453,13 +401,8 @@ def upgrade() -> None: def downgrade() -> None: """Downgrade schema.""" # ### commands auto generated by Alembic - please adjust! ### - # Drop the live-pointer FKs first so the release tables can be dropped below. + # Drop the live-pointer FK first so hook_releases can be dropped below. op.drop_constraint("fk_hooks_live_release_id", "hooks", type_="foreignkey") - op.drop_constraint("fk_ingesters_live_release_id", "ingesters", type_="foreignkey") - op.drop_index("idx_ingester_releases_ingester_version", table_name="ingester_releases") - op.drop_table("ingester_releases") - op.drop_index("idx_ingesters_schema_id", table_name="ingesters") - op.drop_table("ingesters") op.drop_index("idx_hook_runs_release", table_name="hook_runs") op.drop_table("hook_runs") op.drop_index("ix_role_assignments_user_id", table_name="role_assignments") diff --git a/server/migrations/versions/e1a7d20c4f9b_add_ingester_registry.py b/server/migrations/versions/e1a7d20c4f9b_add_ingester_registry.py new file mode 100644 index 00000000..006a95dd --- /dev/null +++ b/server/migrations/versions/e1a7d20c4f9b_add_ingester_registry.py @@ -0,0 +1,108 @@ +"""add ingester registry: identities, releases, run snapshot + +Ingester identity + immutable versioned releases (#180 §1), mirroring the hook +registry, plus ``ingest_runs.release_id`` — the release a run resolved at start, +snapshotted for provenance. Incremental revision (live archives upgrade in +place). + +Greenfield provenance: ``release_id`` is NOT NULL — every run is traceable to +the exact code that fetched its records. Pre-registry runs carry no release to +point at, so the upgrade clears ``ingest_runs``: runs are operational state and +re-runnable (re-ingestion re-creates records with full provenance), not archival +data. + +Revision ID: e1a7d20c4f9b +Revises: 44a8e3799b97 +Create Date: 2026-08-14 + +""" + +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects import postgresql + +# revision identifiers, used by Alembic. +revision: str = "e1a7d20c4f9b" +down_revision: Union[str, Sequence[str], None] = "44a8e3799b97" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + """Upgrade schema.""" + op.create_table( + "ingesters", + sa.Column("name", sa.String(length=40), nullable=False), + sa.Column("schema_id", sa.String(), nullable=False), + sa.Column("live_release_id", sa.UUID(), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.PrimaryKeyConstraint("name"), + ) + op.create_index("idx_ingesters_schema_id", "ingesters", ["schema_id"], unique=False) + op.create_table( + "ingester_releases", + sa.Column("id", sa.UUID(), nullable=False), + sa.Column("ingester_name", sa.String(length=40), nullable=False), + sa.Column("version", sa.Integer(), nullable=False), + sa.Column("image", sa.Text(), nullable=False), + sa.Column("digest", sa.Text(), nullable=False), + sa.Column( + "config", + postgresql.JSONB(astext_type=sa.Text()), + server_default=sa.text("'{}'"), + nullable=False, + ), + sa.Column("limits", postgresql.JSONB(astext_type=sa.Text()), nullable=False), + sa.Column("source_ref", sa.Text(), nullable=False), + sa.Column("built_by", sa.Text(), nullable=True), + sa.Column("built_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint(["ingester_name"], ["ingesters.name"]), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("ingester_name", "digest", name="uq_ingester_releases_ingester_digest"), + sa.UniqueConstraint( + "ingester_name", "version", name="uq_ingester_releases_ingester_version" + ), + ) + op.create_index( + "idx_ingester_releases_ingester_version", + "ingester_releases", + ["ingester_name", sa.literal_column("version DESC")], + unique=False, + ) + # Same circular-dependency break as hooks.live_release_id in initial_schema. + op.create_foreign_key( + "fk_ingesters_live_release_id", + "ingesters", + "ingester_releases", + ["live_release_id"], + ["id"], + deferrable=True, + initially="DEFERRED", + ) + # Pre-registry runs have no release to point at; runs are re-runnable + # operational state, so clear them rather than carry a nullable column. + op.execute("DELETE FROM ingest_runs") + op.add_column( + "ingest_runs", + sa.Column("release_id", sa.UUID(), nullable=False), + ) + op.create_foreign_key( + "fk_ingest_runs_release_id", + "ingest_runs", + "ingester_releases", + ["release_id"], + ["id"], + ) + + +def downgrade() -> None: + """Downgrade schema.""" + op.drop_constraint("fk_ingest_runs_release_id", "ingest_runs", type_="foreignkey") + op.drop_column("ingest_runs", "release_id") + op.drop_constraint("fk_ingesters_live_release_id", "ingesters", type_="foreignkey") + op.drop_index("idx_ingester_releases_ingester_version", table_name="ingester_releases") + op.drop_table("ingester_releases") + op.drop_index("idx_ingesters_schema_id", table_name="ingesters") + op.drop_table("ingesters") diff --git a/server/osa/domain/deposition/command/create_convention.py b/server/osa/domain/deposition/command/create_convention.py index 1e32fd3f..716b30f4 100644 --- a/server/osa/domain/deposition/command/create_convention.py +++ b/server/osa/domain/deposition/command/create_convention.py @@ -28,6 +28,7 @@ from osa.domain.shared.model.source import ( IngesterDefinition, IngesterLimits, + IngesterName, IngesterScheduleConfig, InitialRunConfig, ) @@ -94,7 +95,10 @@ class DeployConventionIngester(BaseModel): model_config = ConfigDict(extra="forbid") - name: str # build-fan-out key (cloud); not persisted server-side + # Build fan-out key (cloud) AND, normalised at this boundary, the ingester's + # registry identity (#180 §1). Wire names are free-form (class names, + # declared slugs); the domain only ever sees a canonical IngesterName. + name: str config: dict[str, Any] | None = None limits: IngesterLimits = Field(default_factory=IngesterLimits) schedule: IngesterScheduleConfig | None = None @@ -103,6 +107,7 @@ class DeployConventionIngester(BaseModel): def to_definition(self) -> IngesterDefinition: return IngesterDefinition( + name=IngesterName.from_raw(self.name), image=self.release.image, digest=self.release.digest, config=self.config, diff --git a/server/osa/domain/deposition/service/convention.py b/server/osa/domain/deposition/service/convention.py index 4b8727e6..9876a911 100644 --- a/server/osa/domain/deposition/service/convention.py +++ b/server/osa/domain/deposition/service/convention.py @@ -7,11 +7,13 @@ from osa.domain.deposition.model.docs import ConventionDocs from osa.domain.deposition.model.value import FileRequirements from osa.domain.deposition.port.convention_repository import ConventionRepository +from osa.domain.ingest.service.ingester_registry import IngesterRegistryService from osa.domain.metadata.service.metadata import MetadataService from osa.domain.semantics.model.value import FieldDefinition from osa.domain.semantics.service.schema import SchemaService from osa.domain.shared.error import NotFoundError from osa.domain.shared.event import EventId +from osa.domain.shared.model.hook import OciConfig, OciLimits from osa.domain.shared.model.source import IngesterDefinition from osa.domain.shared.model.srn import ( ConventionSlug, @@ -30,6 +32,7 @@ class ConventionService(Service): schema_service: SchemaService # TODO: replace with a port? metadata_service: MetadataService # TODO: replace with a port? hook_registry: HookRegistryService + ingester_registry: IngesterRegistryService outbox: Outbox async def deploy( @@ -91,6 +94,26 @@ async def deploy( spec.identity.name, spec.runtime, spec.source_ref, built_by ) + # 2b) Ingester: same treatment as hooks (#180 §1) — upsert identity + + # mint release (idempotent on digest, advancing the live pointer). + # Unconditional: name and source_ref are required on the model, so + # every declared ingester carries a provenance chain. + if ingester is not None: + await self.ingester_registry.upsert_identity( + ingester.name, LocalId(created_schema.id.id.root) + ) + await self.ingester_registry.create_release( + ingester.name, + OciConfig( + image=ingester.image, + digest=ingester.digest, + config=ingester.config if ingester.config is not None else {}, + limits=OciLimits(**ingester.limits.model_dump()), + ), + ingester.source_ref, + built_by, + ) + # 3) Convention referencing hooks by name (upsert by slug). convention = Convention( id=slug, diff --git a/server/osa/domain/deposition/util/di/provider.py b/server/osa/domain/deposition/util/di/provider.py index 3cfba6b3..505061ce 100644 --- a/server/osa/domain/deposition/util/di/provider.py +++ b/server/osa/domain/deposition/util/di/provider.py @@ -20,6 +20,7 @@ from osa.domain.deposition.query.list_depositions import ListDepositionsHandler from osa.domain.deposition.query.list_ingesters import ListIngestersHandler from osa.domain.deposition.service.convention import ConventionService +from osa.domain.ingest.service.ingester_registry import IngesterRegistryService from osa.domain.validation.service.hook_registry import HookRegistryService from osa.domain.deposition.service.deposition import DepositionService from osa.domain.metadata.service.metadata import MetadataService @@ -56,6 +57,7 @@ def get_convention_service( schema_service: SchemaService, metadata_service: MetadataService, hook_registry: HookRegistryService, + ingester_registry: IngesterRegistryService, outbox: Outbox, ) -> ConventionService: return ConventionService( @@ -63,6 +65,7 @@ def get_convention_service( schema_service=schema_service, metadata_service=metadata_service, hook_registry=hook_registry, + ingester_registry=ingester_registry, outbox=outbox, ) diff --git a/server/osa/domain/ingest/model/ingest_run.py b/server/osa/domain/ingest/model/ingest_run.py index ceda6e99..e3e7605a 100644 --- a/server/osa/domain/ingest/model/ingest_run.py +++ b/server/osa/domain/ingest/model/ingest_run.py @@ -5,6 +5,7 @@ from enum import StrEnum from typing import NewType +from osa.domain.ingest.model.ingester_release import IngesterReleaseId from osa.domain.shared.error import InvalidStateError from osa.domain.shared.failure import FailureKind from osa.domain.shared.model.aggregate import Aggregate @@ -36,6 +37,10 @@ class IngestRun(Aggregate): id: IngestRunId convention_id: str + # The ingester release this run resolved at start (#180 §1) — snapshotted, + # exactly as validation snapshots hook releases. Required: every run is + # traceable to the exact code that fetched its records. + release_id: IngesterReleaseId status: IngestStatus = IngestStatus.PENDING ingestion_finished: bool = False batches_ingested: int = 0 diff --git a/server/osa/domain/ingest/service/ingest.py b/server/osa/domain/ingest/service/ingest.py index 77f97615..b4df0646 100644 --- a/server/osa/domain/ingest/service/ingest.py +++ b/server/osa/domain/ingest/service/ingest.py @@ -15,6 +15,7 @@ ) from osa.domain.ingest.port.instrumentation import IngestInstrumentation from osa.domain.ingest.port.repository import IngestRunRepository +from osa.domain.ingest.service.ingester_registry import IngesterRegistryService from osa.domain.shared.error import ConflictError, NotFoundError from osa.domain.shared.event import EventId from osa.domain.shared.failure import FailureKind @@ -31,6 +32,7 @@ class IngestService(Service): ingest_repo: IngestRunRepository convention_service: ConventionService + ingester_registry: IngesterRegistryService outbox: Outbox node_domain: Domain instrumentation: IngestInstrumentation @@ -64,12 +66,18 @@ async def start_ingest( code="ingest_already_running", ) + # Resolve the ingester's live release once, at run start, and snapshot + # it on the run (#180 §1) — mirroring how validation snapshots hook + # releases. No live release ⇒ the run is refused (NotFoundError). + release = await self.ingester_registry.require_live_release(convention.ingester.name) + run_id = IngestRunId(str(uuid4())) now = datetime.now(UTC) ingest_run = IngestRun( id=run_id, convention_id=convention_id, + release_id=release.id, status=IngestStatus.PENDING, batch_size=batch_size, limit=limit, diff --git a/server/osa/domain/shared/model/names.py b/server/osa/domain/shared/model/names.py index f45ced7c..7d3d8dc2 100644 --- a/server/osa/domain/shared/model/names.py +++ b/server/osa/domain/shared/model/names.py @@ -16,7 +16,7 @@ from __future__ import annotations import re -from typing import ClassVar +from typing import ClassVar, Self from pydantic import ConfigDict, RootModel, field_validator @@ -50,3 +50,20 @@ def _validate(cls, value: str) -> str: def __str__(self) -> str: return self.root + + @classmethod + def from_raw(cls, raw: str) -> Self: + """Parse a free-form name into a canonical PgName at the boundary. + + Wire names arrive as Python class names (``PDBIngester``) or declared + slugs (``usgs-feed``). Normalisation is deterministic — lowercase, + non-``[a-z0-9]`` runs collapse to a single ``_``, truncated to the cap — + so the same wire name always maps to the same registry identity. + Raises ``ValueError`` when nothing valid remains (e.g. all digits). + """ + # Break CamelCase before lowering so PDBFeed → pdb_feed, not pdbfeed: + # lower/digit→Upper boundaries, then acronym→word (USGSFeed → USGS_Feed). + decamel = re.sub(r"(?<=[a-z0-9])(?=[A-Z])", "_", raw) + decamel = re.sub(r"(?<=[A-Z])(?=[A-Z][a-z])", "_", decamel) + normalised = re.sub(r"[^a-z0-9]+", "_", decamel.lower()).strip("_") + return cls(normalised[:MAX_NAME_LENGTH].rstrip("_")) diff --git a/server/osa/domain/shared/model/source.py b/server/osa/domain/shared/model/source.py index 3192eb62..25c40623 100644 --- a/server/osa/domain/shared/model/source.py +++ b/server/osa/domain/shared/model/source.py @@ -94,14 +94,20 @@ def _record_source_discriminator(v: Any) -> str: class IngesterDefinition(ValueObject): - """Complete specification for an ingester: image reference + config + limits.""" + """Complete specification for an ingester: identity + image + config + limits. + ``name`` is the ingester's registry identity (#180 §1) — a declared ingester + IS an identity, so a nameless one is unrepresentable. ``source_ref`` is the + reproducibility anchor for the build that produced ``image`` (parity with a + hook release's ``source_ref``); a release without provenance is equally + unrepresentable. + """ + + name: IngesterName image: str digest: str config: dict[str, Any] | None = None limits: IngesterLimits = Field(default_factory=IngesterLimits) schedule: IngesterScheduleConfig | None = None initial_run: InitialRunConfig | None = None - # Reproducibility anchor for the build that produced ``image`` (parity with - # a hook's release ``source_ref``). ``None`` for ingesters predating the field. - source_ref: str | None = None + source_ref: str diff --git a/server/osa/infrastructure/ingest/di.py b/server/osa/infrastructure/ingest/di.py index 970cd8bc..4a850a65 100644 --- a/server/osa/infrastructure/ingest/di.py +++ b/server/osa/infrastructure/ingest/di.py @@ -13,6 +13,7 @@ from osa.domain.ingest.port.repository import IngestRunRepository from osa.domain.ingest.port.storage import IngestStoragePort from osa.domain.ingest.service.ingest import IngestService +from osa.domain.ingest.service.ingester_registry import IngesterRegistryService from osa.domain.shared.model.srn import Domain from osa.domain.shared.outbox import Outbox from osa.infrastructure.persistence.adapter.ingest_storage import FilesystemIngestStorage @@ -35,11 +36,16 @@ def get_storage_layout(self, paths: OSAPaths) -> StorageLayout: def get_ingest_repo(self, session: AsyncSession) -> IngestRunRepository: return PostgresIngestRunRepository(session) + # Ingester registry service (#180 §1) — identity + releases for the code + # that fetches raw data, mirroring the hook registry service. + ingester_registry_service = provide(IngesterRegistryService, scope=Scope.UOW) + @provide(scope=Scope.UOW) def get_ingest_service( self, ingest_repo: IngestRunRepository, convention_service: ConventionService, + ingester_registry: IngesterRegistryService, outbox: Outbox, node_domain: Domain, instrumentation: IngestInstrumentation, @@ -47,6 +53,7 @@ def get_ingest_service( return IngestService( ingest_repo=ingest_repo, convention_service=convention_service, + ingester_registry=ingester_registry, outbox=outbox, node_domain=node_domain, instrumentation=instrumentation, diff --git a/server/osa/infrastructure/persistence/di.py b/server/osa/infrastructure/persistence/di.py index 3e1414ec..dc4f5503 100644 --- a/server/osa/infrastructure/persistence/di.py +++ b/server/osa/infrastructure/persistence/di.py @@ -30,6 +30,7 @@ from osa.domain.shared.port.unit_of_work import UnitOfWork from osa.domain.feature.port.feature_store import FeatureStore from osa.domain.validation.port.repository import ValidationRunRepository +from osa.domain.ingest.port.ingester_registry import IngesterRegistry from osa.domain.validation.port.hook_registry import HookRegistry from osa.domain.data.port.data_read_store import ( DataCatalogReadStore, @@ -53,6 +54,9 @@ from osa.infrastructure.persistence.repository.hook_registry import ( PostgresHookRegistry, ) +from osa.infrastructure.persistence.repository.ingester_registry import ( + PostgresIngesterRegistry, +) from osa.infrastructure.persistence.repository.deposition import ( PostgresDepositionRepository, ) @@ -140,6 +144,12 @@ def get_metadata_store(self, engine: AsyncEngine, session: AsyncSession) -> Meta # the live pointer, and hook_runs (record + provenance reads). hook_registry_repo = provide(PostgresHookRegistry, scope=Scope.UOW, provides=HookRegistry) + # Ingester registry (ingest domain — #180 §1): mirrors the hook registry + # for the code that fetches raw data. + ingester_registry_repo = provide( + PostgresIngesterRegistry, scope=Scope.UOW, provides=IngesterRegistry + ) + # Cross-domain readers schema_reader = provide(SchemaReaderAdapter, scope=Scope.UOW, provides=SchemaReader) ontology_reader = provide(OntologyReaderAdapter, scope=Scope.UOW, provides=OntologyReader) diff --git a/server/osa/infrastructure/persistence/repository/ingest.py b/server/osa/infrastructure/persistence/repository/ingest.py index 013df28f..1d2f8f4b 100644 --- a/server/osa/infrastructure/persistence/repository/ingest.py +++ b/server/osa/infrastructure/persistence/repository/ingest.py @@ -14,6 +14,7 @@ RunClosed, RunUpdate, ) +from osa.domain.ingest.model.ingester_release import IngesterReleaseId from osa.domain.ingest.port.repository import IngestRunRepository from osa.domain.shared.error import NotFoundError from osa.domain.shared.failure import FailureKind @@ -37,6 +38,7 @@ async def save(self, ingest_run: IngestRun) -> None: values = { "id": ingest_run.id, "convention_id": ingest_run.convention_id, + "release_id": ingest_run.release_id, "status": ingest_run.status.value, "ingestion_finished": ingest_run.ingestion_finished, "batches_ingested": ingest_run.batches_ingested, @@ -254,6 +256,7 @@ def _row_to_ingest_run(row: dict) -> IngestRun: return IngestRun( id=row["id"], convention_id=row["convention_id"], + release_id=IngesterReleaseId(row["release_id"]), status=IngestStatus(row["status"]), ingestion_finished=row["ingestion_finished"], batches_ingested=row["batches_ingested"], diff --git a/server/osa/infrastructure/persistence/tables.py b/server/osa/infrastructure/persistence/tables.py index 9498f4c4..0a3a0e30 100644 --- a/server/osa/infrastructure/persistence/tables.py +++ b/server/osa/infrastructure/persistence/tables.py @@ -375,6 +375,14 @@ metadata, Column("id", String, primary_key=True), Column("convention_id", String, nullable=False), + # The ingester release resolved at run start (#180 §1) — NOT NULL: every + # run is traceable to the exact code that fetched its records. + Column( + "release_id", + PGUUID(as_uuid=True), + ForeignKey("ingester_releases.id", name="fk_ingest_runs_release_id"), + nullable=False, + ), Column("status", String(32), nullable=False, server_default=text("'pending'")), Column("ingestion_finished", Boolean, nullable=False, server_default=text("false")), Column("batches_ingested", Integer, nullable=False, server_default=text("0")), diff --git a/server/tests/integration/persistence/test_convention_repo.py b/server/tests/integration/persistence/test_convention_repo.py index f6aea231..2535b45d 100644 --- a/server/tests/integration/persistence/test_convention_repo.py +++ b/server/tests/integration/persistence/test_convention_repo.py @@ -18,6 +18,7 @@ from osa.domain.shared.model.source import ( IngesterDefinition, IngesterLimits, + IngesterName, IngesterScheduleConfig, InitialRunConfig, ) @@ -61,6 +62,7 @@ def _make_hook() -> HookName: def _make_ingester() -> IngesterDefinition: return IngesterDefinition( + name=IngesterName("from_pdb"), image="ghcr.io/example/ingester:latest", digest="sha256:def456", runner="oci", @@ -68,6 +70,7 @@ def _make_ingester() -> IngesterDefinition: limits=IngesterLimits(timeout_seconds=7200, memory="8g", cpu="4.0"), schedule=IngesterScheduleConfig(cron="0 2 * * *", limit=500), initial_run=InitialRunConfig(limit=100), + source_ref="git+https://example.com/r@deadbeef", ) diff --git a/server/tests/integration/persistence/test_ingest_repo.py b/server/tests/integration/persistence/test_ingest_repo.py index 627c5f5c..84009178 100644 --- a/server/tests/integration/persistence/test_ingest_repo.py +++ b/server/tests/integration/persistence/test_ingest_repo.py @@ -8,6 +8,7 @@ from datetime import UTC, datetime import pytest +import pytest_asyncio from sqlalchemy.ext.asyncio import AsyncSession from osa.domain.ingest.model.ingest_run import ( @@ -17,15 +18,42 @@ IngestStatus, RunClosed, ) +from osa.domain.ingest.model.ingester_release import IngesterReleaseId from osa.domain.shared.error import NotFoundError from osa.domain.shared.failure import FailureKind +from osa.domain.shared.model.hook import OciConfig +from osa.domain.shared.model.source import IngesterName +from osa.domain.shared.model.srn import LocalId from osa.infrastructure.persistence.repository.ingest import PostgresIngestRunRepository +from osa.infrastructure.persistence.repository.ingester_registry import ( + PostgresIngesterRegistry, +) + +@pytest_asyncio.fixture +async def release_id(pg_session: AsyncSession) -> IngesterReleaseId: + """Seed an ingester + release; ingest_runs.release_id is a NOT NULL FK.""" + registry = PostgresIngesterRegistry(pg_session) + await registry.upsert_identity(IngesterName("from_pdb"), LocalId("proteinschema")) + outcome = await registry.create_release( + IngesterName("from_pdb"), + OciConfig(image="ghcr.io/x:v1", digest="sha256:abc"), + "git+https://example.com/r@deadbeef", + None, + ) + return outcome.release.id -def _make_run(run_id: str = "ing-1", status: IngestStatus = IngestStatus.RUNNING) -> IngestRun: + +def _make_run( + run_id: str = "ing-1", + status: IngestStatus = IngestStatus.RUNNING, + *, + release_id: IngesterReleaseId, +) -> IngestRun: return IngestRun( id=IngestRunId(run_id), convention_id="test-conv", + release_id=release_id, status=status, batch_size=100, batches_ingested=1, @@ -35,9 +63,11 @@ def _make_run(run_id: str = "ing-1", status: IngestStatus = IngestStatus.RUNNING @pytest.mark.asyncio class TestFailureSurfacing: - async def test_record_failure_round_trips_reason_and_kind(self, pg_session: AsyncSession): + async def test_record_failure_round_trips_reason_and_kind( + self, pg_session: AsyncSession, release_id + ): repo = PostgresIngestRunRepository(pg_session) - await repo.save(_make_run("ing-rt")) + await repo.save(_make_run("ing-rt", release_id=release_id)) await repo.record_failure("ing-rt", reason="source exited 3", kind=FailureKind.UPSTREAM) @@ -46,9 +76,9 @@ async def test_record_failure_round_trips_reason_and_kind(self, pg_session: Asyn assert fetched.failure_reason == "source exited 3" assert fetched.failure_kind is FailureKind.UPSTREAM - async def test_record_failure_allows_null_kind(self, pg_session: AsyncSession): + async def test_record_failure_allows_null_kind(self, pg_session: AsyncSession, release_id): repo = PostgresIngestRunRepository(pg_session) - await repo.save(_make_run("ing-null")) + await repo.save(_make_run("ing-null", release_id=release_id)) await repo.record_failure("ing-null", reason="retries exhausted", kind=None) @@ -58,11 +88,11 @@ async def test_record_failure_allows_null_kind(self, pg_session: AsyncSession): assert fetched.failure_kind is None async def test_record_failure_does_not_clobber_an_existing_reason( - self, pg_session: AsyncSession + self, pg_session: AsyncSession, release_id ): """An abort sets the reason first; a later batch give-up must not overwrite it.""" repo = PostgresIngestRunRepository(pg_session) - await repo.save(_make_run("ing-clobber")) + await repo.save(_make_run("ing-clobber", release_id=release_id)) aborted = await repo.abort( "ing-clobber", @@ -94,9 +124,11 @@ class TestTerminalRunGuard: non-terminal status, mirroring `abort`. """ - async def test_increment_failed_is_noop_on_terminal_run(self, pg_session: AsyncSession): + async def test_increment_failed_is_noop_on_terminal_run( + self, pg_session: AsyncSession, release_id + ): repo = PostgresIngestRunRepository(pg_session) - await repo.save(_make_run("ing-tf")) + await repo.save(_make_run("ing-tf", release_id=release_id)) await repo.abort( "ing-tf", reason="bad image", @@ -112,9 +144,11 @@ async def test_increment_failed_is_noop_on_terminal_run(self, pg_session: AsyncS assert fetched.batches_failed == 0 assert fetched.status is IngestStatus.FAILED - async def test_increment_completed_is_noop_on_terminal_run(self, pg_session: AsyncSession): + async def test_increment_completed_is_noop_on_terminal_run( + self, pg_session: AsyncSession, release_id + ): repo = PostgresIngestRunRepository(pg_session) - await repo.save(_make_run("ing-tc")) + await repo.save(_make_run("ing-tc", release_id=release_id)) await repo.abort( "ing-tc", reason="rbac", kind=FailureKind.RBAC, completed_at=datetime.now(UTC) ) @@ -127,17 +161,19 @@ async def test_increment_completed_is_noop_on_terminal_run(self, pg_session: Asy assert fetched.batches_completed == 0 assert fetched.published_count == 0 - async def test_increment_still_works_on_a_running_run(self, pg_session: AsyncSession): + async def test_increment_still_works_on_a_running_run( + self, pg_session: AsyncSession, release_id + ): """Sanity: the guard doesn't break the normal (non-terminal) path.""" repo = PostgresIngestRunRepository(pg_session) - await repo.save(_make_run("ing-live")) + await repo.save(_make_run("ing-live", release_id=release_id)) result = await repo.increment_failed("ing-live") assert isinstance(result, Applied) assert result.run.batches_failed == 1 - async def test_increment_on_missing_run_raises(self, pg_session: AsyncSession): + async def test_increment_on_missing_run_raises(self, pg_session: AsyncSession, release_id): """A genuinely-missing run is exceptional — not a RunClosed no-op.""" repo = PostgresIngestRunRepository(pg_session) with pytest.raises(NotFoundError): @@ -148,9 +184,11 @@ async def test_increment_on_missing_run_raises(self, pg_session: AsyncSession): class TestMarkBatchIngested: """Idempotent batch-ingested marker with GREATEST semantics (#160).""" - async def test_advances_counter_to_batch_index_plus_one(self, pg_session: AsyncSession): + async def test_advances_counter_to_batch_index_plus_one( + self, pg_session: AsyncSession, release_id + ): repo = PostgresIngestRunRepository(pg_session) - run = _make_run("ing-mb1") + run = _make_run("ing-mb1", release_id=release_id) run.batches_ingested = 0 await repo.save(run) @@ -160,9 +198,9 @@ async def test_advances_counter_to_batch_index_plus_one(self, pg_session: AsyncS assert result.run.batches_ingested == 1 assert result.run.ingestion_finished is False - async def test_idempotent_on_repeat_with_same_index(self, pg_session: AsyncSession): + async def test_idempotent_on_repeat_with_same_index(self, pg_session: AsyncSession, release_id): repo = PostgresIngestRunRepository(pg_session) - run = _make_run("ing-mb2") + run = _make_run("ing-mb2", release_id=release_id) run.batches_ingested = 0 await repo.save(run) @@ -175,10 +213,12 @@ async def test_idempotent_on_repeat_with_same_index(self, pg_session: AsyncSessi assert first.run.batches_ingested == 4 assert second.run.batches_ingested == 4 - async def test_never_rolls_back_a_counter_that_ran_ahead(self, pg_session: AsyncSession): + async def test_never_rolls_back_a_counter_that_ran_ahead( + self, pg_session: AsyncSession, release_id + ): """A late delivery for an earlier batch must not lower the counter.""" repo = PostgresIngestRunRepository(pg_session) - run = _make_run("ing-mb3") + run = _make_run("ing-mb3", release_id=release_id) run.batches_ingested = 5 await repo.save(run) @@ -187,9 +227,9 @@ async def test_never_rolls_back_a_counter_that_ran_ahead(self, pg_session: Async assert isinstance(result, Applied) assert result.run.batches_ingested == 5 # GREATEST(5, 2) = 5 - async def test_ingestion_finished_latches_true(self, pg_session: AsyncSession): + async def test_ingestion_finished_latches_true(self, pg_session: AsyncSession, release_id): repo = PostgresIngestRunRepository(pg_session) - run = _make_run("ing-mb4") + run = _make_run("ing-mb4", release_id=release_id) run.batches_ingested = 0 await repo.save(run) @@ -202,9 +242,9 @@ async def test_ingestion_finished_latches_true(self, pg_session: AsyncSession): assert isinstance(again, Applied) assert again.run.ingestion_finished is True - async def test_noop_on_terminal_run(self, pg_session: AsyncSession): + async def test_noop_on_terminal_run(self, pg_session: AsyncSession, release_id): repo = PostgresIngestRunRepository(pg_session) - await repo.save(_make_run("ing-mb5")) + await repo.save(_make_run("ing-mb5", release_id=release_id)) await repo.abort( "ing-mb5", reason="bad image", @@ -219,7 +259,23 @@ async def test_noop_on_terminal_run(self, pg_session: AsyncSession): assert fetched is not None assert fetched.batches_ingested == 1 # _make_run seeds 1; abort didn't advance it - async def test_missing_run_raises(self, pg_session: AsyncSession): + async def test_missing_run_raises(self, pg_session: AsyncSession, release_id): repo = PostgresIngestRunRepository(pg_session) with pytest.raises(NotFoundError): await repo.mark_batch_ingested("does-not-exist", 0, ingestion_finished=False) + + +@pytest.mark.asyncio +class TestReleaseSnapshot: + """ingest_runs.release_id (#180 §1) round-trips against real PG. + + Greenfield: the column is NOT NULL with a real FK — there is no + release-less run to round-trip.""" + + async def test_release_id_round_trips_with_fk(self, pg_session: AsyncSession, release_id): + repo = PostgresIngestRunRepository(pg_session) + await repo.save(_make_run("ing-rel", release_id=release_id)) + + loaded = await repo.get("ing-rel") + assert loaded is not None + assert loaded.release_id == release_id diff --git a/server/tests/integration/test_bulk_publish_dual_write.py b/server/tests/integration/test_bulk_publish_dual_write.py index bb57be50..4f26043d 100644 --- a/server/tests/integration/test_bulk_publish_dual_write.py +++ b/server/tests/integration/test_bulk_publish_dual_write.py @@ -35,10 +35,14 @@ from osa.domain.semantics.service.schema import SchemaService from osa.domain.shared.model.source import DepositionSource from osa.domain.shared.model.srn import ConventionSlug, Domain, SchemaIdentifier +from osa.domain.ingest.service.ingester_registry import IngesterRegistryService from osa.domain.validation.service.hook_registry import HookRegistryService from osa.infrastructure.persistence.metadata_store import PostgresMetadataStore from osa.infrastructure.persistence.repository.convention import PostgresConventionRepository from osa.infrastructure.persistence.repository.hook_registry import PostgresHookRegistry +from osa.infrastructure.persistence.repository.ingester_registry import ( + PostgresIngesterRegistry, +) from osa.infrastructure.persistence.repository.ontology import PostgresOntologyRepository from osa.infrastructure.persistence.repository.record import PostgresRecordRepository from osa.infrastructure.persistence.repository.schema import PostgresSemanticsSchemaRepository @@ -78,6 +82,7 @@ async def _register_convention( schema_service=schema_service, metadata_service=metadata_service, hook_registry=HookRegistryService(registry=PostgresHookRegistry(pg_session)), + ingester_registry=IngesterRegistryService(registry=PostgresIngesterRegistry(pg_session)), outbox=AsyncMock(), ) # Bundled deploy: schema + typed metadata table + convention, one txn. diff --git a/server/tests/unit/application/workflow/test_process_batch.py b/server/tests/unit/application/workflow/test_process_batch.py index 339f55bf..2686cc02 100644 --- a/server/tests/unit/application/workflow/test_process_batch.py +++ b/server/tests/unit/application/workflow/test_process_batch.py @@ -21,6 +21,7 @@ IngestBatchPublished, NextBatchRequested, ) +from osa.domain.ingest.model.ingester_release import IngesterReleaseId from osa.domain.ingest.model.ingest_run import ( Applied, IngestRun, @@ -32,7 +33,7 @@ from osa.domain.shared.event import EventId from osa.domain.shared.failure import DecisionKind, FailureKind, FailurePolicy, RuntimeFailure from osa.domain.shared.model.hook import HookName, OciConfig, OciLimits, TableFeatureSpec -from osa.domain.shared.model.source import IngesterDefinition +from osa.domain.shared.model.source import IngesterDefinition, IngesterName from osa.domain.shared.model.srn import ConventionSlug, Domain, LocalId, RecordSRN, RecordVersion from osa.domain.shared.model.workflow import StageOutcome, WorkflowName, WorkflowStage from osa.domain.shared.port.ingester_runner import IngesterOutput @@ -120,6 +121,7 @@ def _make_run( return IngestRun( id=IngestRunId("run-1"), convention_id="test-conv", + release_id=IngesterReleaseId(uuid4()), status=status, batch_size=100, batches_ingested=batches_ingested, @@ -242,7 +244,12 @@ def _make_handler( convention_service = AsyncMock() conv = AsyncMock() conv.hooks = [HookName(n) for n in hook_names] - conv.ingester = IngesterDefinition(image="ghcr.io/x/ing:v1", digest="sha256:abc") + conv.ingester = IngesterDefinition( + name=IngesterName("from_pdb"), + image="ghcr.io/x/ing:v1", + digest="sha256:abc", + source_ref="git+https://example.com/r@deadbeef", + ) conv.id = ConventionSlug.parse("test-conv") convention_service.get_convention.return_value = conv diff --git a/server/tests/unit/domain/deposition/test_convention_service.py b/server/tests/unit/domain/deposition/test_convention_service.py index 7ed28acd..c7b7d0ed 100644 --- a/server/tests/unit/domain/deposition/test_convention_service.py +++ b/server/tests/unit/domain/deposition/test_convention_service.py @@ -88,6 +88,7 @@ def _make_service( schema_service=mock_schema_service, metadata_service=AsyncMock(), hook_registry=hook_registry or AsyncMock(), + ingester_registry=AsyncMock(), outbox=outbox or AsyncMock(), ) diff --git a/server/tests/unit/domain/deposition/test_convention_service_v2.py b/server/tests/unit/domain/deposition/test_convention_service_v2.py index 96342e80..e002cdb3 100644 --- a/server/tests/unit/domain/deposition/test_convention_service_v2.py +++ b/server/tests/unit/domain/deposition/test_convention_service_v2.py @@ -3,6 +3,7 @@ from unittest.mock import AsyncMock import pytest +from pydantic import ValidationError as PydanticValidationError from tests.factories import make_convention_docs from osa.domain.deposition.event.convention_registered import ConventionRegistered @@ -18,7 +19,7 @@ OciConfig, TableFeatureSpec, ) -from osa.domain.shared.model.source import IngesterDefinition +from osa.domain.shared.model.source import IngesterDefinition, IngesterName from osa.domain.shared.model.srn import ( ConventionSlug, SchemaId, @@ -70,12 +71,16 @@ def _make_hook_deploy(name: str = "detect_pockets") -> HookDeploy: ) -def _make_ingester_def() -> IngesterDefinition: - return IngesterDefinition( +def _make_ingester_def(**overrides) -> IngesterDefinition: + kwargs = dict( + name=IngesterName("rcsb_pdb"), image="osa-sources/rcsb-pdb:latest", digest="sha256:abc123", config={"email": "test@example.com", "batch_size": 100}, + source_ref="git+https://example.com/rcsb-pdb@abc", ) + kwargs.update(overrides) + return IngesterDefinition(**kwargs) def _make_service( @@ -83,6 +88,7 @@ def _make_service( schema_service: AsyncMock | None = None, outbox: AsyncMock | None = None, hook_registry: AsyncMock | None = None, + ingester_registry: AsyncMock | None = None, ) -> ConventionService: """Create a ConventionService with mock deps.""" mock_schema_service = schema_service or AsyncMock() @@ -99,6 +105,7 @@ def _make_service( schema_service=mock_schema_service, metadata_service=AsyncMock(), hook_registry=hook_registry or AsyncMock(), + ingester_registry=ingester_registry or AsyncMock(), outbox=outbox or AsyncMock(), ) @@ -199,3 +206,58 @@ async def test_deploy_without_source_still_emits_event(self): emitted = outbox.append.call_args[0][0] assert isinstance(emitted, ConventionRegistered) assert emitted.convention_id == result.id + + +class TestDeployMintsIngesterRelease: + """Deploy wires the ingester registry (#180 §1): a declared ingester gets an + identity + release minted in the same deploy, unconditionally — greenfield, + no unnamed/pre-registry path. Name and source_ref are required at the model + boundary, so the only deploy-time behavior to test is the minting itself.""" + + def _ingester(self, **overrides) -> IngesterDefinition: + kwargs = dict( + name=IngesterName("from_pdb"), + image="ghcr.io/example/ingester:v1", + digest="sha256:abc123", + source_ref="git+https://example.com/repo@deadbeef", + ) + kwargs.update(overrides) + return IngesterDefinition(**kwargs) + + @pytest.mark.asyncio + async def test_declared_ingester_mints_identity_and_release(self): + registry = AsyncMock() + service = _make_service(ingester_registry=registry) + await _deploy(service, ingester=self._ingester(), built_by="ci@example") + + registry.upsert_identity.assert_awaited_once() + name_arg, schema_arg = registry.upsert_identity.await_args[0] + assert name_arg == IngesterName("from_pdb") + assert schema_arg.root == "testschema12345678" + + registry.create_release.assert_awaited_once() + rel_args = registry.create_release.await_args + assert rel_args[0][0] == IngesterName("from_pdb") + runtime = rel_args[0][1] + assert runtime.image == "ghcr.io/example/ingester:v1" + assert runtime.digest == "sha256:abc123" + assert rel_args[0][2] == "git+https://example.com/repo@deadbeef" + assert rel_args[0][3] == "ci@example" + + @pytest.mark.asyncio + async def test_no_ingester_mints_nothing(self): + registry = AsyncMock() + service = _make_service(ingester_registry=registry) + await _deploy(service, ingester=None) + registry.upsert_identity.assert_not_awaited() + registry.create_release.assert_not_awaited() + + def test_nameless_ingester_is_unrepresentable(self): + """Greenfield: a declared ingester IS an identity — pydantic rejects the + nameless form at construction, so no service-level guard can exist.""" + with pytest.raises(PydanticValidationError): + IngesterDefinition(image="ghcr.io/x:v1", digest="sha256:def456") + + def test_ingester_without_source_ref_is_unrepresentable(self): + with pytest.raises(PydanticValidationError): + self._ingester(source_ref=None) diff --git a/server/tests/unit/domain/deposition/test_deploy_convention_dto.py b/server/tests/unit/domain/deposition/test_deploy_convention_dto.py index f3346fc8..4ce4ff99 100644 --- a/server/tests/unit/domain/deposition/test_deploy_convention_dto.py +++ b/server/tests/unit/domain/deposition/test_deploy_convention_dto.py @@ -137,10 +137,38 @@ def test_ingester_to_definition_regathers() -> None: assert idef.source_ref == "git:1" # provenance carried onto the ingester -def test_ingester_name_accepted_but_not_persisted() -> None: - # `name` is the cloud build-fan-out key; the internal ingester has none. +def test_ingester_name_normalised_into_registry_identity() -> None: + # `name` is the cloud build-fan-out key AND, normalised at this boundary, + # the ingester's registry identity (#180 §1). Wire names are free-form + # (class names, declared slugs); the domain only sees a canonical PgName. idef = DeployConventionIngester.model_validate(_ingester(name="anything")).to_definition() - assert "name" not in idef.model_dump() + assert idef.name.root == "anything" + + +def test_ingester_name_normalisation_is_deterministic() -> None: + for wire, expected in [ + ("USGSFeed", "usgs_feed"), + ("usgs-feed", "usgs_feed"), + ("PDBIngester", "pdb_ingester"), + ]: + idef = DeployConventionIngester.model_validate(_ingester(name=wire)).to_definition() + assert idef.name.root == expected, wire + + +def test_ingester_name_is_required() -> None: + # A declared ingester IS an identity — a nameless one is unrepresentable. + body = _ingester() + del body["name"] + with pytest.raises(ValidationError): + DeployConventionIngester.model_validate(body) + + +def test_ingester_source_ref_is_required() -> None: + # A release without provenance is the thing the registry exists to prevent. + body = _ingester() + del body["release"]["source_ref"] + with pytest.raises(ValidationError): + DeployConventionIngester.model_validate(body) def test_ingester_rejects_flat_image() -> None: diff --git a/server/tests/unit/domain/ingest/test_get_ingestion.py b/server/tests/unit/domain/ingest/test_get_ingestion.py index 4fe4e39b..e535b90b 100644 --- a/server/tests/unit/domain/ingest/test_get_ingestion.py +++ b/server/tests/unit/domain/ingest/test_get_ingestion.py @@ -4,10 +4,12 @@ from unittest.mock import AsyncMock, MagicMock import pytest +from uuid import uuid4 from osa.domain.auth.model.principal import Principal from osa.domain.auth.model.role import Role from osa.domain.auth.model.value import ProviderIdentity, UserId +from osa.domain.ingest.model.ingester_release import IngesterReleaseId from osa.domain.ingest.model.ingest_run import IngestRun, IngestRunId, IngestStatus from osa.domain.ingest.query.get_ingestion import GetIngestion, GetIngestionHandler from osa.domain.shared.error import NotFoundError @@ -28,6 +30,7 @@ def _make_run(**overrides) -> IngestRun: defaults = { "id": IngestRunId("run-1"), "convention_id": "test-conv", + "release_id": IngesterReleaseId(uuid4()), "status": IngestStatus.RUNNING, "batch_size": 100, "started_at": _T0, @@ -90,6 +93,7 @@ async def test_service_raises_not_found_for_missing_run(self) -> None: service = IngestService( ingest_repo=repo, convention_service=AsyncMock(), + ingester_registry=AsyncMock(), outbox=AsyncMock(), node_domain=Domain("localhost"), instrumentation=MagicMock(), diff --git a/server/tests/unit/domain/ingest/test_ingest_run.py b/server/tests/unit/domain/ingest/test_ingest_run.py index 1d1ac815..6446849d 100644 --- a/server/tests/unit/domain/ingest/test_ingest_run.py +++ b/server/tests/unit/domain/ingest/test_ingest_run.py @@ -3,7 +3,9 @@ from datetime import UTC, datetime import pytest +from uuid import uuid4 +from osa.domain.ingest.model.ingester_release import IngesterReleaseId from osa.domain.ingest.model.ingest_run import IngestRun, IngestStatus from osa.domain.shared.error import InvalidStateError from osa.domain.shared.failure import FailureKind @@ -13,6 +15,7 @@ def _make_run(**overrides) -> IngestRun: defaults = { "id": "test-run-id", "convention_id": "urn:osa:localhost:conv:test-conv@1.0.0", + "release_id": IngesterReleaseId(uuid4()), "status": IngestStatus.PENDING, "started_at": datetime.now(UTC), } diff --git a/server/tests/unit/domain/ingest/test_ingest_service.py b/server/tests/unit/domain/ingest/test_ingest_service.py index afa8d46b..50c625f7 100644 --- a/server/tests/unit/domain/ingest/test_ingest_service.py +++ b/server/tests/unit/domain/ingest/test_ingest_service.py @@ -2,13 +2,15 @@ from datetime import UTC, datetime from unittest.mock import AsyncMock, MagicMock +from uuid import uuid4 import pytest from osa.domain.ingest.model.ingest_run import Applied, IngestStatus, RunClosed +from osa.domain.ingest.model.ingester_release import IngesterReleaseId from osa.domain.ingest.service.ingest import IngestService from osa.domain.shared.error import ConflictError, NotFoundError -from osa.domain.shared.model.source import IngesterDefinition +from osa.domain.shared.model.source import IngesterDefinition, IngesterName from osa.domain.shared.model.srn import Domain @@ -35,8 +37,10 @@ def _make_convention(*, has_ingester: bool = True): conv.srn = "test-conv" conv.ingester = ( IngesterDefinition( + name=IngesterName("from_pdb"), image="ghcr.io/example/ingester:v1", digest="sha256:abc123", + source_ref="git+https://example.com/repo@deadbeef", ) if has_ingester else None @@ -44,11 +48,18 @@ def _make_convention(*, has_ingester: bool = True): return conv +def _make_release() -> MagicMock: + release = MagicMock() + release.id = IngesterReleaseId(uuid4()) + return release + + def _make_service( *, convention=None, running_ingest=None, convention_not_found: bool = False, + ingester_registry: AsyncMock | None = None, ) -> IngestService: ingest_repo = AsyncMock() ingest_repo.get_running_for_convention.return_value = running_ingest @@ -62,9 +73,14 @@ def _make_service( outbox = AsyncMock() + if ingester_registry is None: + ingester_registry = AsyncMock() + ingester_registry.require_live_release.return_value = _make_release() + return IngestService( ingest_repo=ingest_repo, convention_service=convention_service, + ingester_registry=ingester_registry, outbox=outbox, node_domain=Domain("localhost"), instrumentation=RecordingIngestInstrumentation(), @@ -151,6 +167,7 @@ async def test_pending_run_marked_running_and_saved(self) -> None: run = IngestRun( id=IngestRunId("run-1"), convention_id="test-conv", + release_id=IngesterReleaseId(uuid4()), status=IngestStatus.PENDING, started_at=datetime.now(UTC), ) @@ -170,6 +187,7 @@ async def test_running_run_is_noop(self) -> None: run = IngestRun( id=IngestRunId("run-1"), convention_id="test-conv", + release_id=IngesterReleaseId(uuid4()), status=IngestStatus.RUNNING, started_at=datetime.now(UTC), ) @@ -258,6 +276,7 @@ async def test_aborts_via_atomic_repo_update(self) -> None: run=IngestRun( id=IngestRunId("run-1"), convention_id="test-conv", + release_id=IngesterReleaseId(uuid4()), status=IngestStatus.FAILED, failure_reason="Image pull failed: 401", failure_kind=FailureKind.IMAGE_PULL, @@ -454,6 +473,7 @@ async def test_completion_emits_run_finished_once(self) -> None: run = IngestRun( id=IngestRunId("run-1"), convention_id="test-conv", + release_id=IngesterReleaseId(uuid4()), status=IngestStatus.RUNNING, ingestion_finished=True, batches_ingested=1, @@ -512,6 +532,7 @@ async def test_abort_emits_run_finished_failed(self) -> None: run=IngestRun( id=IngestRunId("run-1"), convention_id="test-conv", + release_id=IngesterReleaseId(uuid4()), status=IngestStatus.FAILED, started_at=_dt.now(UTC), ) @@ -532,3 +553,36 @@ async def test_abort_noop_emits_no_run_finished(self) -> None: await service.abort_run(IngestRunId("run-1"), reason="x", kind=FailureKind.RBAC) assert service.instrumentation.runs_finished == [] + + +class TestStartIngestSnapshotsRelease: + """start_ingest resolves the ingester's live release and snapshots it on the + run (#180 §1) — mirroring how validation snapshots hook releases. Greenfield: + resolution is unconditional; a convention whose ingester has no live release + cannot start a run.""" + + @pytest.mark.asyncio + async def test_stamps_release_id_on_the_run(self): + release = _make_release() + registry = AsyncMock() + registry.require_live_release.return_value = release + service = _make_service(ingester_registry=registry) + + run = await service.start_ingest("test-conv") + + registry.require_live_release.assert_awaited_once_with(IngesterName("from_pdb")) + assert run.release_id == release.id + saved = service.ingest_repo.save.await_args[0][0] + assert saved.release_id == release.id + + @pytest.mark.asyncio + async def test_no_live_release_refuses_the_run(self): + registry = AsyncMock() + registry.require_live_release.side_effect = NotFoundError( + "No live release for ingester 'from_pdb'" + ) + service = _make_service(ingester_registry=registry) + + with pytest.raises(NotFoundError): + await service.start_ingest("test-conv") + service.ingest_repo.save.assert_not_awaited() diff --git a/server/tests/unit/infrastructure/k8s/test_k8s_ingester_runner.py b/server/tests/unit/infrastructure/k8s/test_k8s_ingester_runner.py index 8c4a5dfc..4fa37bfe 100644 --- a/server/tests/unit/infrastructure/k8s/test_k8s_ingester_runner.py +++ b/server/tests/unit/infrastructure/k8s/test_k8s_ingester_runner.py @@ -8,7 +8,7 @@ from osa.config import K8sConfig from osa.domain.shared.failure import FailureKind, RuntimeFailure -from osa.domain.shared.model.source import IngesterDefinition, IngesterLimits +from osa.domain.shared.model.source import IngesterDefinition, IngesterLimits, IngesterName from osa.domain.shared.model.srn import ConventionSlug from osa.domain.shared.port.ingester_runner import IngesterInputs from osa.infrastructure.k8s.ingester_runner import K8sIngesterRunner @@ -25,10 +25,12 @@ def _make_ingester( config: dict[str, Any] | None = None, ) -> IngesterDefinition: return IngesterDefinition( + name=IngesterName("from_pdb"), image=image, digest=digest, config=config, limits=IngesterLimits(timeout_seconds=timeout, memory=memory, cpu=cpu), + source_ref="git+https://example.com/r@deadbeef", ) From 009df8e585c93468ce05d8b7bb2c584cd51127d2 Mon Sep 17 00:00:00 2001 From: Rory Byrne Date: Sat, 15 Aug 2026 12:32:22 +0100 Subject: [PATCH 4/5] fix(ingest): release idempotency by definition equality; reject overlong names MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two Greptile P1s on the wiring commit, both verified real: - Digest-only dedupe let a config/limits/source_ref change silently retain the old release while the convention ran the new settings — the snapshotted release_id would describe something that didn't run. create_release now compares the full definition (image, digest, config, limits, source_ref; built_by excluded) against the LIVE release: byte-identical redeploy is a no-op, anything else mints vN+1. The (name, digest) unique constraint is dropped — a config-only redeploy legitimately mints a new release with the same digest. - PgName.from_raw silently truncated overlong names at 40 chars, aliasing distinct long names into one registry identity. It now rejects with the limit named; the deploy DTO validates at parse (422). Case/separator unification (PDBFeed = pdb-feed) is documented as the intended semantics, not a collision. --- .../e1a7d20c4f9b_add_ingester_registry.py | 4 +- .../deposition/command/create_convention.py | 8 +- .../domain/ingest/port/ingester_registry.py | 16 ++- .../ingest/service/ingester_registry.py | 6 +- server/osa/domain/shared/model/names.py | 19 +++- .../repository/ingester_registry.py | 40 ++++--- .../osa/infrastructure/persistence/tables.py | 4 +- .../test_ingester_registry_repo.py | 104 ++++++++++++++++++ .../deposition/test_deploy_convention_dto.py | 7 ++ 9 files changed, 178 insertions(+), 30 deletions(-) create mode 100644 server/tests/integration/persistence/test_ingester_registry_repo.py diff --git a/server/migrations/versions/e1a7d20c4f9b_add_ingester_registry.py b/server/migrations/versions/e1a7d20c4f9b_add_ingester_registry.py index 006a95dd..9c713378 100644 --- a/server/migrations/versions/e1a7d20c4f9b_add_ingester_registry.py +++ b/server/migrations/versions/e1a7d20c4f9b_add_ingester_registry.py @@ -60,7 +60,9 @@ def upgrade() -> None: sa.Column("built_at", sa.DateTime(timezone=True), nullable=False), sa.ForeignKeyConstraint(["ingester_name"], ["ingesters.name"]), sa.PrimaryKeyConstraint("id"), - sa.UniqueConstraint("ingester_name", "digest", name="uq_ingester_releases_ingester_digest"), + # No (name, digest) unique constraint: a config-only redeploy mints a + # new release with the same digest (idempotency is definition-equality + # against the live release, decided in the adapter). sa.UniqueConstraint( "ingester_name", "version", name="uq_ingester_releases_ingester_version" ), diff --git a/server/osa/domain/deposition/command/create_convention.py b/server/osa/domain/deposition/command/create_convention.py index 716b30f4..42028642 100644 --- a/server/osa/domain/deposition/command/create_convention.py +++ b/server/osa/domain/deposition/command/create_convention.py @@ -1,7 +1,7 @@ from datetime import datetime from typing import Any -from pydantic import BaseModel, ConfigDict, Field, model_validator +from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator from osa.domain.auth.model.principal import Principal from osa.domain.deposition.model.deploy import HookDeploy @@ -105,6 +105,12 @@ class DeployConventionIngester(BaseModel): initial_run: InitialRunConfig | None = None release: DeployConventionRelease + @field_validator("name") + @classmethod + def _name_normalises(cls, v: str) -> str: + IngesterName.from_raw(v) # overlong/unnormalisable → 422 here, not deep in deploy + return v + def to_definition(self) -> IngesterDefinition: return IngesterDefinition( name=IngesterName.from_raw(self.name), diff --git a/server/osa/domain/ingest/port/ingester_registry.py b/server/osa/domain/ingest/port/ingester_registry.py index 63223769..986228e7 100644 --- a/server/osa/domain/ingest/port/ingester_registry.py +++ b/server/osa/domain/ingest/port/ingester_registry.py @@ -45,12 +45,16 @@ async def create_release( ) -> IngesterReleaseOutcome: """Mint the next release for an existing ingester and advance live. - Idempotent on ``(name, digest)``: re-submitting an existing digest - returns the existing release without minting a version or moving the - pointer, with ``created == False``. Version assignment and pointer - advance happen under a row lock on the ``ingesters`` row so concurrent - submitters serialize, and ``created`` is decided under that same lock so - it is race-free. + Idempotent on **definition equality with the live release**: a redeploy + of exactly what is live (image, digest, config, limits, source_ref — + ``built_by`` excluded) returns the live release without minting a + version or moving the pointer, with ``created == False``. Any + difference mints vN+1, so a run's ``release_id`` always describes + exactly what was deployed — dedupe on digest alone would silently + retain stale config. Version assignment and pointer advance happen + under a row lock on the ``ingesters`` row so concurrent submitters + serialize, and ``created`` is decided under that same lock so it is + race-free. """ ... diff --git a/server/osa/domain/ingest/service/ingester_registry.py b/server/osa/domain/ingest/service/ingester_registry.py index a791cfb8..06c7a5a7 100644 --- a/server/osa/domain/ingest/service/ingester_registry.py +++ b/server/osa/domain/ingest/service/ingester_registry.py @@ -35,7 +35,11 @@ async def create_release( source_ref: str, built_by: str | None = None, ) -> IngesterReleaseOutcome: - """Mint vN+1 for an existing ingester (idempotent on digest); advance live.""" + """Mint vN+1 for an existing ingester; advance live. + + Idempotent on definition-equality with the live release: only a + redeploy of exactly what is live is a no-op (see the port docstring). + """ return await self.registry.create_release(name, runtime, source_ref, built_by) async def set_live(self, name: IngesterName, version: int) -> Ingester: diff --git a/server/osa/domain/shared/model/names.py b/server/osa/domain/shared/model/names.py index 7d3d8dc2..16952c4b 100644 --- a/server/osa/domain/shared/model/names.py +++ b/server/osa/domain/shared/model/names.py @@ -57,13 +57,24 @@ def from_raw(cls, raw: str) -> Self: Wire names arrive as Python class names (``PDBIngester``) or declared slugs (``usgs-feed``). Normalisation is deterministic — lowercase, - non-``[a-z0-9]`` runs collapse to a single ``_``, truncated to the cap — - so the same wire name always maps to the same registry identity. - Raises ``ValueError`` when nothing valid remains (e.g. all digits). + non-``[a-z0-9]`` runs collapse to a single ``_`` — so the same wire + name always maps to the same registry identity. Spelling variants of + the same words (``PDBFeed``, ``pdb-feed``, ``pdb_feed``) unifying into + one identity is the intended semantics, not a collision. + + Overlong names are REJECTED, never truncated: silent truncation would + alias distinct long names into one identity. Raises ``ValueError`` when + the normalised name exceeds the cap or nothing valid remains (e.g. all + digits). """ # Break CamelCase before lowering so PDBFeed → pdb_feed, not pdbfeed: # lower/digit→Upper boundaries, then acronym→word (USGSFeed → USGS_Feed). decamel = re.sub(r"(?<=[a-z0-9])(?=[A-Z])", "_", raw) decamel = re.sub(r"(?<=[A-Z])(?=[A-Z][a-z])", "_", decamel) normalised = re.sub(r"[^a-z0-9]+", "_", decamel.lower()).strip("_") - return cls(normalised[:MAX_NAME_LENGTH].rstrip("_")) + if len(normalised) > MAX_NAME_LENGTH: + raise ValueError( + f"{cls.kind} {raw!r} normalises to {len(normalised)} chars; " + f"the maximum is {MAX_NAME_LENGTH}. Use a shorter name." + ) + return cls(normalised) diff --git a/server/osa/infrastructure/persistence/repository/ingester_registry.py b/server/osa/infrastructure/persistence/repository/ingester_registry.py index e421c1aa..f362f72f 100644 --- a/server/osa/infrastructure/persistence/repository/ingester_registry.py +++ b/server/osa/infrastructure/persistence/repository/ingester_registry.py @@ -116,25 +116,33 @@ async def create_release( locked = await self.session.execute( select(ingesters_table).where(ingesters_table.c.name == name.root).with_for_update() ) - if locked.mappings().first() is None: + ingester_row = locked.mappings().first() + if ingester_row is None: raise NotFoundError(f"Ingester not found: {name.root}") - # Idempotency on (ingester_name, digest): return the existing release, no - # new version, pointer unchanged. Decided under the row lock, so `created` - # is race-free under concurrent identical submissions. - duplicate = await self.session.execute( - select(ingester_releases_table).where( - and_( - ingester_releases_table.c.ingester_name == name.root, - ingester_releases_table.c.digest == runtime.digest, - ) - ) - ) - duplicate_row = duplicate.mappings().first() - if duplicate_row is not None: - return IngesterReleaseOutcome( - release=self._to_release(dict(duplicate_row)), created=False + # Idempotency by DEFINITION EQUALITY against the live release: only a + # redeploy of exactly what is already live is a no-op. Any difference — + # image, digest, config, limits, source_ref — mints a new version, so a + # run's release_id always describes exactly what was deployed (digest + # alone would silently retain stale config). ``built_by`` is excluded: + # who built it does not change what it is. Decided under the row lock, + # so ``created`` is race-free under concurrent identical submissions. + live_id = ingester_row["live_release_id"] + if live_id is not None: + live = await self.session.execute( + select(ingester_releases_table).where(ingester_releases_table.c.id == live_id) ) + live_row = live.mappings().first() + if live_row is not None and ( + live_row["image"] == runtime.image + and live_row["digest"] == runtime.digest + and live_row["config"] == runtime.config + and live_row["limits"] == runtime.limits.model_dump() + and live_row["source_ref"] == source_ref + ): + return IngesterReleaseOutcome( + release=self._to_release(dict(live_row)), created=False + ) highest_version = await self.session.scalar( select(func.coalesce(func.max(ingester_releases_table.c.version), 0)).where( diff --git a/server/osa/infrastructure/persistence/tables.py b/server/osa/infrastructure/persistence/tables.py index 0a3a0e30..b8d5719b 100644 --- a/server/osa/infrastructure/persistence/tables.py +++ b/server/osa/infrastructure/persistence/tables.py @@ -525,7 +525,9 @@ Column("built_by", Text, nullable=True), Column("built_at", DateTime(timezone=True), nullable=False), UniqueConstraint("ingester_name", "version", name="uq_ingester_releases_ingester_version"), - UniqueConstraint("ingester_name", "digest", name="uq_ingester_releases_ingester_digest"), + # NB: deliberately NO (name, digest) unique constraint — a config-only + # redeploy mints a new release with the same digest (idempotency is by + # definition-equality against the live release, decided in the adapter). ) Index( diff --git a/server/tests/integration/persistence/test_ingester_registry_repo.py b/server/tests/integration/persistence/test_ingester_registry_repo.py new file mode 100644 index 00000000..bf317292 --- /dev/null +++ b/server/tests/integration/persistence/test_ingester_registry_repo.py @@ -0,0 +1,104 @@ +"""Integration tests for PostgresIngesterRegistry release idempotency. + +Releases are immutable and idempotent by *definition equality against the live +release* — not by digest alone. A redeploy that changes anything the release +records (image, digest, config, limits, source_ref) mints a new version, so a +run's ``release_id`` always describes exactly what was deployed; only a +byte-identical redeploy is a no-op. (``built_by`` is excluded: who built it +does not change what it is.) +""" + +import pytest +from sqlalchemy.ext.asyncio import AsyncSession + +from osa.domain.shared.model.hook import OciConfig, OciLimits +from osa.domain.shared.model.source import IngesterName +from osa.domain.shared.model.srn import LocalId +from osa.infrastructure.persistence.repository.ingester_registry import ( + PostgresIngesterRegistry, +) + +_NAME = IngesterName("from_pdb") +_SOURCE_REF = "git+https://example.com/r@deadbeef" + + +def _runtime(*, digest: str = "sha256:abc", batch_size: int = 100) -> OciConfig: + return OciConfig( + image="ghcr.io/x:v1", + digest=digest, + config={"batch_size": batch_size}, + limits=OciLimits(), + ) + + +async def _seeded_registry(pg_session: AsyncSession) -> PostgresIngesterRegistry: + registry = PostgresIngesterRegistry(pg_session) + await registry.upsert_identity(_NAME, LocalId("proteinschema")) + return registry + + +@pytest.mark.asyncio +class TestReleaseIdempotency: + async def test_identical_redeploy_is_a_noop(self, pg_session: AsyncSession): + registry = await _seeded_registry(pg_session) + first = await registry.create_release(_NAME, _runtime(), _SOURCE_REF, "ci@example") + + again = await registry.create_release(_NAME, _runtime(), _SOURCE_REF, "ci@example") + + assert first.created is True + assert again.created is False + assert again.release.id == first.release.id + ingester = await registry.get_ingester(_NAME) + assert ingester is not None and ingester.live_release_id == first.release.id + + async def test_config_only_change_mints_new_release(self, pg_session: AsyncSession): + """Same digest, different config: the release must describe what runs — + digest-only dedupe would silently retain stale config (Greptile P1).""" + registry = await _seeded_registry(pg_session) + first = await registry.create_release(_NAME, _runtime(batch_size=100), _SOURCE_REF, None) + + second = await registry.create_release(_NAME, _runtime(batch_size=500), _SOURCE_REF, None) + + assert second.created is True + assert second.release.id != first.release.id + assert second.release.version == first.release.version + 1 + assert second.release.runtime.digest == first.release.runtime.digest + ingester = await registry.get_ingester(_NAME) + assert ingester is not None and ingester.live_release_id == second.release.id + + async def test_digest_change_mints_new_release(self, pg_session: AsyncSession): + registry = await _seeded_registry(pg_session) + first = await registry.create_release( + _NAME, _runtime(digest="sha256:abc"), _SOURCE_REF, None + ) + + second = await registry.create_release( + _NAME, _runtime(digest="sha256:def"), _SOURCE_REF, None + ) + + assert second.created is True + assert second.release.version == first.release.version + 1 + + async def test_built_by_alone_does_not_mint(self, pg_session: AsyncSession): + """Who built it doesn't change what it is.""" + registry = await _seeded_registry(pg_session) + first = await registry.create_release(_NAME, _runtime(), _SOURCE_REF, "alice@example") + + again = await registry.create_release(_NAME, _runtime(), _SOURCE_REF, "bob@example") + + assert again.created is False + assert again.release.id == first.release.id + + async def test_rollback_then_identical_redeploy_matches_live(self, pg_session: AsyncSession): + """Idempotency compares against the LIVE release, honouring rollbacks: + after set_live(v1), redeploying v1's definition is a no-op even though + v2 exists.""" + registry = await _seeded_registry(pg_session) + first = await registry.create_release(_NAME, _runtime(batch_size=100), _SOURCE_REF, None) + await registry.create_release(_NAME, _runtime(batch_size=500), _SOURCE_REF, None) + await registry.set_live(_NAME, first.release.version) + + again = await registry.create_release(_NAME, _runtime(batch_size=100), _SOURCE_REF, None) + + assert again.created is False + assert again.release.id == first.release.id diff --git a/server/tests/unit/domain/deposition/test_deploy_convention_dto.py b/server/tests/unit/domain/deposition/test_deploy_convention_dto.py index 4ce4ff99..eccae6be 100644 --- a/server/tests/unit/domain/deposition/test_deploy_convention_dto.py +++ b/server/tests/unit/domain/deposition/test_deploy_convention_dto.py @@ -155,6 +155,13 @@ def test_ingester_name_normalisation_is_deterministic() -> None: assert idef.name.root == expected, wire +def test_ingester_name_too_long_is_rejected_not_truncated() -> None: + # Silent truncation would alias distinct long names into one registry + # identity (Greptile P1 on #208). Greenfield: reject, name the limit. + with pytest.raises(ValidationError): + DeployConventionIngester.model_validate(_ingester(name="x" * 41)) + + def test_ingester_name_is_required() -> None: # A declared ingester IS an identity — a nameless one is unrepresentable. body = _ingester() From d869f42c4ab5ff68f8f1ee85a2b61d97eb7b1050 Mon Sep 17 00:00:00 2001 From: Rory Byrne Date: Sat, 15 Aug 2026 19:40:28 +0100 Subject: [PATCH 5/5] fix(validation): hook release idempotency by definition equality MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixes #217. Mirrors the ingester registry exactly (009df8e), so both registries share one rule before #218 collapses them into the Function substrate: - create_release compares the full definition (image, digest, config, limits, source_ref; built_by excluded) against the LIVE release — byte-identical redeploy is a no-op, anything else mints vN+1. Hook runs execute from the live release, so the old digest-only dedupe made config-only redeploys silently never take effect. - uq_hook_releases_hook_digest dropped (incremental migration b47f9c2e8a31 — metadata-only, safe on populated tables): a config-only redeploy legitimately mints a new release with the same digest. - Port/service/deploy docstrings updated; integration suite mirrored from the ingester side (identical no-op, config-only mints, rollback compares against live). --- ...47f9c2e8a31_drop_hook_release_digest_uq.py | 41 +++++++ .../domain/deposition/service/convention.py | 7 +- .../domain/validation/port/hook_registry.py | 16 ++- .../validation/service/hook_registry.py | 5 +- .../persistence/repository/hook_registry.py | 34 ++++-- .../osa/infrastructure/persistence/tables.py | 4 +- .../test_hook_registry_idempotency.py | 111 ++++++++++++++++++ 7 files changed, 194 insertions(+), 24 deletions(-) create mode 100644 server/migrations/versions/b47f9c2e8a31_drop_hook_release_digest_uq.py create mode 100644 server/tests/integration/persistence/test_hook_registry_idempotency.py diff --git a/server/migrations/versions/b47f9c2e8a31_drop_hook_release_digest_uq.py b/server/migrations/versions/b47f9c2e8a31_drop_hook_release_digest_uq.py new file mode 100644 index 00000000..71562d49 --- /dev/null +++ b/server/migrations/versions/b47f9c2e8a31_drop_hook_release_digest_uq.py @@ -0,0 +1,41 @@ +"""drop hook_releases (name, digest) unique constraint + +Hook release idempotency changes from digest-only to definition-equality +against the live release (#217): a config-only redeploy now mints a new +release with the same digest, so the digest uniqueness constraint must go. +Mirrors the ingester registry, which shipped without the constraint. + +Safe on populated tables: dropping a unique constraint is metadata-only. + +Revision ID: b47f9c2e8a31 +Revises: e1a7d20c4f9b +Create Date: 2026-08-15 + +""" + +from typing import Sequence, Union + +from alembic import op + +# revision identifiers, used by Alembic. +revision: str = "b47f9c2e8a31" +down_revision: Union[str, Sequence[str], None] = "e1a7d20c4f9b" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + """Upgrade schema.""" + op.drop_constraint("uq_hook_releases_hook_digest", "hook_releases", type_="unique") + + +def downgrade() -> None: + """Downgrade schema. + + NB: recreating the constraint fails if any hook has minted multiple + releases with one digest since upgrading — expected, and correct: those + rows are legitimate under the new semantics. + """ + op.create_unique_constraint( + "uq_hook_releases_hook_digest", "hook_releases", ["hook_name", "digest"] + ) diff --git a/server/osa/domain/deposition/service/convention.py b/server/osa/domain/deposition/service/convention.py index 9876a911..cffeb649 100644 --- a/server/osa/domain/deposition/service/convention.py +++ b/server/osa/domain/deposition/service/convention.py @@ -58,7 +58,8 @@ async def deploy( are unversioned and mutable, so re-declaring the same state is a no-op and a differing declaration updates the convention in place — no caller version, no conflict path. The schema (versioned, immutable) and each hook release - (idempotent on digest) are reused when already present. Feature-table + (idempotent on definition-equality with the live release) are reused + when already present. Feature-table creation is now inlined at the deploy command handler (#160, decision 9); the ``ConventionRegistered`` event this method still appends is audit-only (no subscribers) — it lives on for the ``/events`` changefeed. @@ -87,7 +88,7 @@ async def deploy( ) # 2) Hooks: upsert each identity (reject a differing contract) + mint its - # release (idempotent on digest, advancing the live pointer). + # release (idempotent on definition-equality, advancing the live pointer). for spec in hooks: await self.hook_registry.upsert_identity(spec.identity.name, spec.identity.feature) await self.hook_registry.create_release( @@ -95,7 +96,7 @@ async def deploy( ) # 2b) Ingester: same treatment as hooks (#180 §1) — upsert identity + - # mint release (idempotent on digest, advancing the live pointer). + # mint release (idempotent on definition-equality, advancing the live pointer). # Unconditional: name and source_ref are required on the model, so # every declared ingester carries a provenance chain. if ingester is not None: diff --git a/server/osa/domain/validation/port/hook_registry.py b/server/osa/domain/validation/port/hook_registry.py index 86538635..5a60299b 100644 --- a/server/osa/domain/validation/port/hook_registry.py +++ b/server/osa/domain/validation/port/hook_registry.py @@ -38,12 +38,16 @@ async def create_release( ) -> ReleaseOutcome: """Mint the next release for an existing hook and advance the live pointer. - Idempotent on ``(name, digest)``: re-submitting an existing digest returns - the existing release without minting a new version or moving the pointer - (FR-006/R5), with ``ReleaseOutcome.created == False``. Version assignment + - pointer advance happen under a row lock on the ``hooks`` row so concurrent - submitters serialize (FR-009/R7); ``created`` is decided under that same - lock so it is race-free. + Idempotent on **definition equality with the live release** (#217): a + redeploy of exactly what is live (image, digest, config, limits, + source_ref — ``built_by`` excluded) returns the live release without + minting a version or moving the pointer, with + ``ReleaseOutcome.created == False``. Any difference mints vN+1 — hook + runs execute from the live release, so digest-only dedupe would make a + config-only redeploy silently never take effect. Version assignment + + pointer advance happen under a row lock on the ``hooks`` row so + concurrent submitters serialize (FR-009/R7); ``created`` is decided + under that same lock so it is race-free. """ ... diff --git a/server/osa/domain/validation/service/hook_registry.py b/server/osa/domain/validation/service/hook_registry.py index 06ddf241..0dbcf55c 100644 --- a/server/osa/domain/validation/service/hook_registry.py +++ b/server/osa/domain/validation/service/hook_registry.py @@ -30,7 +30,10 @@ async def create_release( source_ref: str, built_by: str | None = None, ) -> ReleaseOutcome: - """Mint vN+1 for an existing hook (idempotent on digest); advance live. + """Mint vN+1 for an existing hook; advance live. + + Idempotent on definition-equality with the live release (#217) — only + a redeploy of exactly what is live is a no-op; see the port docstring. Returns a :class:`ReleaseOutcome` whose ``created`` flag (decided under the registry's row lock) distinguishes a new version from an idempotent diff --git a/server/osa/infrastructure/persistence/repository/hook_registry.py b/server/osa/infrastructure/persistence/repository/hook_registry.py index 13cb85d0..4d53b79e 100644 --- a/server/osa/infrastructure/persistence/repository/hook_registry.py +++ b/server/osa/infrastructure/persistence/repository/hook_registry.py @@ -123,20 +123,28 @@ async def create_release( if hook_row is None: raise NotFoundError(f"Hook not found: {name}") - # Idempotency on (hook_name, digest): return the existing release, no - # new version, pointer unchanged (R5). Decided under the row lock, so - # `created` is race-free under concurrent identical submissions. - dup = await self.session.execute( - select(hook_releases_table).where( - and_( - hook_releases_table.c.hook_name == name.root, - hook_releases_table.c.digest == runtime.digest, - ) + # Idempotency by DEFINITION EQUALITY against the live release (#217), + # mirroring the ingester registry: only a redeploy of exactly what is + # live is a no-op. Any difference — image, digest, config, limits, + # source_ref — mints vN+1; hook runs execute from the live release, so + # digest-only dedupe made config-only redeploys silently never take + # effect. ``built_by`` is excluded: who built it does not change what + # it is. Decided under the row lock, so ``created`` is race-free under + # concurrent identical submissions. + live_id = hook_row["live_release_id"] + if live_id is not None: + live = await self.session.execute( + select(hook_releases_table).where(hook_releases_table.c.id == live_id) ) - ) - dup_row = dup.mappings().first() - if dup_row is not None: - return ReleaseOutcome(release=self._to_release(dict(dup_row)), created=False) + live_row = live.mappings().first() + if live_row is not None and ( + live_row["image"] == runtime.image + and live_row["digest"] == runtime.digest + and live_row["config"] == runtime.config + and live_row["limits"] == runtime.limits.model_dump() + and live_row["source_ref"] == source_ref + ): + return ReleaseOutcome(release=self._to_release(dict(live_row)), created=False) max_version = await self.session.scalar( select(func.coalesce(func.max(hook_releases_table.c.version), 0)).where( diff --git a/server/osa/infrastructure/persistence/tables.py b/server/osa/infrastructure/persistence/tables.py index b8d5719b..7e7a54a2 100644 --- a/server/osa/infrastructure/persistence/tables.py +++ b/server/osa/infrastructure/persistence/tables.py @@ -471,7 +471,9 @@ Column("built_by", Text, nullable=True), Column("built_at", DateTime(timezone=True), nullable=False), UniqueConstraint("hook_name", "version", name="uq_hook_releases_hook_version"), - UniqueConstraint("hook_name", "digest", name="uq_hook_releases_hook_digest"), + # NB: deliberately NO (name, digest) unique constraint (#217) — a config-only + # redeploy mints a new release with the same digest (idempotency is + # definition-equality against the live release, decided in the adapter). ) Index( diff --git a/server/tests/integration/persistence/test_hook_registry_idempotency.py b/server/tests/integration/persistence/test_hook_registry_idempotency.py new file mode 100644 index 00000000..b5a7eeb8 --- /dev/null +++ b/server/tests/integration/persistence/test_hook_registry_idempotency.py @@ -0,0 +1,111 @@ +"""Integration tests for PostgresHookRegistry release idempotency (#217). + +Mirrors ``test_ingester_registry_repo.py::TestReleaseIdempotency`` — the two +registries must share one rule until #218 collapses them into the Function +substrate. Releases are idempotent by *definition equality against the live +release*, not by digest: a config-only redeploy previously returned the stale +release and silently never took effect (hook runs execute from the live +release), which #217 fixes. +""" + +import pytest +from sqlalchemy.ext.asyncio import AsyncSession + +from osa.domain.shared.model.hook import ( + ColumnDef, + HookName, + OciConfig, + OciLimits, + TableFeatureSpec, +) +from osa.infrastructure.persistence.repository.hook_registry import PostgresHookRegistry + +_NAME = HookName("detect_pockets") +_SOURCE_REF = "git+https://example.com/pocketeer@deadbeef" + + +def _feature() -> TableFeatureSpec: + return TableFeatureSpec( + cardinality="many", + columns=[ColumnDef(name="score", json_type="number", required=True)], + ) + + +def _runtime(*, digest: str = "sha256:abc", batch_size: int = 100) -> OciConfig: + return OciConfig( + image="ghcr.io/x/pocketeer:v1", + digest=digest, + config={"batch_size": batch_size}, + limits=OciLimits(), + ) + + +async def _seeded_registry(pg_session: AsyncSession) -> PostgresHookRegistry: + registry = PostgresHookRegistry(pg_session) + await registry.upsert_identity(_NAME, _feature()) + return registry + + +@pytest.mark.asyncio +class TestHookReleaseIdempotency: + async def test_identical_redeploy_is_a_noop(self, pg_session: AsyncSession): + registry = await _seeded_registry(pg_session) + first = await registry.create_release(_NAME, _runtime(), _SOURCE_REF, "ci@example") + + again = await registry.create_release(_NAME, _runtime(), _SOURCE_REF, "ci@example") + + assert first.created is True + assert again.created is False + assert again.release.id == first.release.id + hook = await registry.get_hook(_NAME) + assert hook is not None and hook.live_release_id == first.release.id + + async def test_config_only_change_mints_new_release(self, pg_session: AsyncSession): + """#217: same digest, different config MUST mint — hook runs execute from + the live release, so digest-only dedupe made config redeploys silently + never take effect.""" + registry = await _seeded_registry(pg_session) + first = await registry.create_release(_NAME, _runtime(batch_size=100), _SOURCE_REF, None) + + second = await registry.create_release(_NAME, _runtime(batch_size=500), _SOURCE_REF, None) + + assert second.created is True + assert second.release.id != first.release.id + assert second.release.version == first.release.version + 1 + assert second.release.runtime.digest == first.release.runtime.digest + hook = await registry.get_hook(_NAME) + assert hook is not None and hook.live_release_id == second.release.id + + async def test_digest_change_mints_new_release(self, pg_session: AsyncSession): + registry = await _seeded_registry(pg_session) + first = await registry.create_release( + _NAME, _runtime(digest="sha256:abc"), _SOURCE_REF, None + ) + + second = await registry.create_release( + _NAME, _runtime(digest="sha256:def"), _SOURCE_REF, None + ) + + assert second.created is True + assert second.release.version == first.release.version + 1 + + async def test_built_by_alone_does_not_mint(self, pg_session: AsyncSession): + registry = await _seeded_registry(pg_session) + first = await registry.create_release(_NAME, _runtime(), _SOURCE_REF, "alice@example") + + again = await registry.create_release(_NAME, _runtime(), _SOURCE_REF, "bob@example") + + assert again.created is False + assert again.release.id == first.release.id + + async def test_rollback_then_identical_redeploy_matches_live(self, pg_session: AsyncSession): + """Idempotency compares against the LIVE release, honouring rollbacks.""" + registry = await _seeded_registry(pg_session) + first = await registry.create_release(_NAME, _runtime(batch_size=100), _SOURCE_REF, None) + await registry.create_release(_NAME, _runtime(batch_size=500), _SOURCE_REF, None) + await registry.set_live(_NAME, first.release.version) + + again = await registry.create_release(_NAME, _runtime(batch_size=100), _SOURCE_REF, None) + + assert again.created is False + assert again.release.id == first.release.id