feat(ingest): ingester identity + versioned releases (raw-ingestion provenance) - #208
Conversation
|
rorybyrne
changed the title
04e234f to
1c30e84
Compare
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.
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.
1c30e84 to
0344c7c
Compare
Wires the #208 registry into its callers, completing #180 §1's chain record → ingest_run → ingester_release: - Deploy: a named ingester gets upsert_identity + create_release (idempotent on digest), mirroring hooks. The wire name (build-fan-out key, previously discarded) is normalised into the registry identity at the DTO boundary via PgName.from_raw — free-form class names and slugs map deterministically onto the PG-safe charset. Named-without-source_ref is a 422: a named ingester promises provenance. - Run start: start_ingest resolves the live release and snapshots it on ingest_runs.release_id, exactly as ValidationService pins hook releases. - Unnamed ingesters (pre-registry SDKs) deploy and run as before, with a structured warning instead of a provenance chain. The registry DDL moves out of initial_schema into an incremental revision (e1a7d20c4f9b): live archives now exist, so migrations must upgrade in place. Verified: upgrade with pre-existing ingest_runs rows (release_id backfills as NULL), alembic check zero-drift, downgrade preserves data.
Greptile SummaryThe PR adds immutable, versioned ingester releases and snapshots the selected release onto each ingest run for raw-ingestion provenance.
Confidence Score: 5/5The PR appears safe to merge. No blocking failure remains.
|
| Filename | Overview |
|---|---|
| server/osa/infrastructure/persistence/repository/ingester_registry.py | Implements transactionally serialized identity upserts, definition-based release minting, live-pointer updates, and release resolution without a remaining eligible defect. |
| server/osa/domain/ingest/service/ingest.py | Resolves the current ingester release when starting a run and persists its identifier as the provenance snapshot. |
| server/osa/domain/deposition/service/convention.py | Registers the declared ingester identity and release alongside convention deployment through the new registry service. |
| server/osa/domain/shared/model/names.py | Centralizes PostgreSQL-safe nominal names and rejects normalized values exceeding the identifier budget. |
| server/migrations/versions/e1a7d20c4f9b_add_ingester_registry.py | Adds ingester registry storage and a required release foreign key on ingest runs as an incremental migration. |
| server/migrations/versions/b47f9c2e8a31_drop_hook_release_digest_uq.py | Removes the hook digest uniqueness constraint so definition-changing releases may share an image digest. |
Flowchart
%%{init: {'theme': 'neutral'}}%%
flowchart LR
D[Convention deploy] --> I[Ingester identity]
D --> R[Immutable ingester release]
R --> L[Live release pointer]
S[Start ingest] --> L
L --> IR[Ingest run release snapshot]
IR --> P[Raw-ingestion provenance]
Reviews (5): Last reviewed commit: "fix(validation): hook release idempotenc..." | Re-trigger Greptile
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
| 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 | ||
| ) |
There was a problem hiding this comment.
When a named ingester is redeployed with the same digest but changed image reference, config, limits, source ref, or build identity, this branch retains the old immutable release while the mutable convention stores and executes the new settings. The resulting ingest run points to release metadata that does not describe what actually ran.
Knowledge Base Used:
There was a problem hiding this comment.
Confirmed real — fixed in 009df8e. create_release now decides idempotency by definition equality against the live release (image, digest, config, limits, source_ref; built_by excluded — who built it doesn't change what it is): a byte-identical redeploy is a no-op, any difference mints vN+1, so a run's release_id always describes exactly what was deployed. The (name, digest) unique constraint is dropped (migration edited in place — unreleased) since a config-only redeploy now legitimately mints a new release with the same digest. Covered by 5 new integration tests in tests/integration/persistence/test_ingester_registry_repo.py, including the config-only-change case this comment describes and the rollback-then-redeploy edge. Note: the hook registry has the same digest-only semantics (this code deliberately mirrored it) — follow-up tracked separately rather than expanding this PR into the validation domain.
| # 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("_")) |
There was a problem hiding this comment.
When distinct wire names normalize identically, such as PDBFeed and pdb_feed, or share the same normalized first 40 characters, they become one global ingester identity. Same-schema conventions then silently share a live release pointer and can record another convention's release as provenance, while different-schema conventions receive a false schema conflict.
Knowledge Base Used:
There was a problem hiding this comment.
Half-confirmed — the real half fixed in 009df8e. Silent truncation is gone: PgName.from_raw now rejects names that normalise to >40 chars (ValueError naming the limit), surfaced as a 422 at DTO parse via a field_validator — so the truncation-collision vector is closed (test: test_ingester_name_too_long_is_rejected_not_truncated). The case/separator unification (PDBFeed ≡ pdb-feed ≡ pdb_feed) is retained deliberately: deterministic normalisation means the same name spelled differently IS the same registry identity — now documented as intended semantics in the from_raw docstring. Cross-schema aliasing already fails loudly via ingester_schema_conflict; same-schema unification of genuinely distinct ingesters with colliding spellings is accepted as out-of-scope here (ingester identity scoping is part of the Convention-removal rework, #180 PR 3).
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.
0344c7c to
f25e175
Compare
|
@greptile I force-pushed different commits to the original review |
…ong names 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.
| # 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) |
There was a problem hiding this comment.
When a convention is redeployed after an ingest run starts but before a batch executes, ProcessBatch reloads and runs the convention's current ingester definition instead of the release stored on the run, causing records—or different batches in one run—to be attributed to a release that did not produce them.
Knowledge Base Used:
There was a problem hiding this comment.
Confirmed real — and deliberately not patched in this PR. The gap is structural: execution reads the mutable convention.ingester while provenance records the immutable release, and pinning ProcessBatch to the snapshot would be dead code within weeks — #218 (just filed) replaces per-batch orchestration entirely with long-lived streaming sessions that pin their release at session start; invocations (container executions, including crash restarts) inherit the session's release, which closes this misattribution window by construction rather than by patch. Scope of this PR stays: identity + release minting + run-start snapshot. The snapshot column is not wasted interim work — it is the same field the session model reads. Until #218 lands, the window is: convention redeployed mid-run ⇒ later batches run the new definition while run.release_id names the old release — acknowledged in the PR body's scope note.
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).
1 participant
Ingester identity and immutable versioned releases (#180 §1), mirroring the hook
registry — now fully wired, greenfield. Raw-data fetching gets the same
"which exact code did this" provenance that hooks have had since #145:
What this lands
Registry substrate (first two commits):
Ingesteraggregate (name, target schema, live-release pointer) andIngesterRelease(immutable, monotonic per name). Idempotency is bydefinition equality against the live release, decided inside the row
lock (201 vs 200 without a racy pre-check): a byte-identical redeploy is a
no-op; any change — including config-only, same digest — mints vN+1, so a
run's
release_idalways describes exactly what was deployed.built_byis excluded (who built it doesn't change what it is). Reuses
OciConfig:"how a container runs" is one shape.
IngesterRegistryport, service, Postgres adapter.PgNamefactored out ofHookName/FeatureName(identical regex/validatorcopied twice, about to be three) — the 40-char cap and FK-identifier
arithmetic live in one place, now with
from_raw()normalising free-formwire names (
USGSFeed→usgs_feed) deterministically at the DTO boundary.Wiring (third commit, greenfield — no backcompat surface anywhere):
IngesterDefinition.nameand.source_refare required. A declaredingester IS an identity and promises provenance; the nameless/unversioned
forms are unrepresentable, so no "older SDK" guard paths exist to test or
maintain.
Rebinding an ingester to a different schema is a
ConflictError(itspublished records belong to the original schema).
start_ingestresolves the live release once and snapshots it:ingest_runs.release_idis NOT NULL with a real FK. No live release ⇒the run is refused with a 404 naming the ingester.
Migration
e1a7d20c4f9b— an incremental revision (policy change: livearchives now exist, so schema changes ship as upgrade-in-place revisions;
nothing folds into
initial_schemaanymore). It creates the two registrytables and clears
ingest_runsbefore adding the NOT NULL column: pre-registryruns have no release to point at, and runs are re-runnable operational state,
not archival data. Re-ingest after upgrading.
Also in this PR: the hook registry twin (Fixes #217)
Verifying the digest-dedupe P1 revealed the hook registry has the identical defect — worse there, since hook runs execute from the live release: a config-only hook redeploy silently never took effect. Fixed here as an exact mirror (
d869f42) so both registries share one rule before #218 unifies them: definition-equality idempotency,uq_hook_releases_hook_digestdropped via incremental migrationb47f9c2e8a31, mirrored integration suite.Known window (closed structurally by #218)
Until #218's streaming sessions land, batch execution still reads
convention.ingester(the mutable definition) while the run snapshots the release: a convention redeployed mid-run means later batches run the new definition whilerun.release_idnames the old release. Deliberately not patched here — #218 pins execution to the session's release by construction, and aProcessBatchpatch would be deleted with the batch machinery itself.Scope correction (supersedes earlier draft note)
The earlier draft framed record-metadata provenance as "metadata-as-feature"
(
feature.metadata_*tables) and flagged a blocking identifier-budget problem(
metadata_<schema>_<version>at 76 chars vs the 40/63 caps). Both areresolved by decision: metadata keeps its own
metadata.<slug>_v<major>tables (they already exist) with analogous handling — run provenance lands on
them later via the harmoniser's run, mirroring feature tables without being
one. The PG schema namespace carries the
metadataprefix, so the identifierbudget holds and no naming change is needed. The three-link provenance picture
stands; this PR is the raw-data link.
Nullability audit (greenfield rule: no Optional/nullable for backcompat)
IngesterDefinition.name,.source_refingest_runs.release_idConvention.ingester| None— deposition-only conventions are real (dies with #180 PR 3)ingesters.live_release_idhooks.live_release_id(tighten both together or neither)ingester_releases.built_byhook_releases; genuinely unknown for unauthenticated local deploysVerification
live release, name normalisation determinism, unrepresentability of
nameless/unversioned ingesters).
initial → instance_statistics → ingester_registry; release-id FK round-trip).44a8e3799b97, seeded with apre-registry
ingest_runsrow, thenalembic upgrade head— succeeds, runscleared,
release_idNOT NULL present,alembic checkzero drift.ruff+ty check osaclean.Refs #180, #164