0pt/docs/P060_INGESTION_PLANE.md

0pt ingestion plane (P060–P068)

Written 2026-09-04. Every claim here is backed by a test path or a recorded command; anything not verified says so. Read STATUS.md's "P060-P068 ingestion plane" section for the classification table and the honest-gaps list — this document is the mechanism, that one is the status.

The shape

  Source Registry            Connector                Ingestion Runner
  (P060: who, what,          (P061: nine              (P062: one envelope,
   under what rights,         operations, one          one durable checkpoint)
   on what schedule)          conformance suite)
        │                          │                          │
        │  classification gate     │  discover / validate     │  read ─┐
        │  blocks activation       │  connect / read          │        │
        ▼                          ▼  resume / backfill       ▼        ▼
   opt_sources                 records ───────────────▶  IngestionEnvelope
   opt_source_rights                                          │
   opt_source_authorization                                   ├─▶ P063 schema gate
   opt_source_schedule                                        ├─▶ P064 quality gate
   opt_source_schemas                                         ├─▶ P068 ledger (peek)
   opt_ingestion_checkpoints                                  ├─▶ SINK (caller-supplied)
                                                              ├─▶ P068 ledger (commit)
                                                              ├─▶ checkpoint (commit)
                                                              └─▶ P067 lineage edge

P060 — the registry, and the gate

A source now carries: source_id, tenant_id, project_id, source_type, owner_id, an authorization record, a license/purpose rights record, provenance (how the data is obtained), a schedule, a versioned schema, a durable checkpoint, and a status.

status starts at pending. It was active before — both repository implementations wrote input.status ?? "active" — which meant an unclassified source went live the moment it was registered. activateSource is now the only path to active, and it refuses unless ALL of these are on record:

• an owner_id (distinct from created_by: who is accountable, not who typed it),
• an unexpired authorization granted by an accountable actor (svc_*/system/automation are rejected — a source may not authorize itself),
• rights with a license_id and at least one allowed_purpose,
• provenance with a sourcing method.

Schedule, schema and checkpoint describe HOW a source runs, not WHETHER it may, so they are deliberately not part of the gate.

The default is enforced in two places on purpose: in the repositories, and as ALTER TABLE opt_sources ALTER COLUMN status SET DEFAULT 'pending', so a raw INSERT from psql cannot produce an unclassified-but-active source either (proved in the realinfra suite).

Routes: GET/POST /api/0pt/sources/{id}/activate (GET returns the classification without changing anything), GET/PUT …/authorization, GET/PUT …/schedule.

P061 — the connector contract

Nine operations on one interface: validate (pure, no IO, fail-closed), connect (idempotent), discover, health (reports, never throws), read (honours limit exactly), checkpoint, resume, backfill (bounded — an open-ended range is refused), disconnect (idempotent, safe after an error).

src/connectors/conformance.ts is a plain async function, not a test file, so the same suite runs from tests, from an operator script against a real configured source, and (later) from a pre-activation check. There is no "skip" verdict: a connector that cannot honestly serve a check FAILS it.

Two heterogeneous connectors pass it:

	HttpJsonConnector	JsonlFileConnector
transport	network, real HTTP	local filesystem
cursor	opaque server token, with a cursor_field fallback	line offset it computes itself
exhaustion	server-reported	exact (the file is finite)
health	HTTP round trip	stat
backfill bounds	server-defined (sequence/timestamp)	line numbers
CDC deletes	no — it does not claim cdc	yes, via _op

The cursor_field fallback exists because the conformance suite caught a real defect: an API that stops returning next_cursor at the end of a page left the connector with nothing to resume from, so resume() silently restarted at the beginning.

P062 — one envelope, one durable checkpoint

IngestionEnvelope is the single shape everything downstream sees: mode (batch/stream/cdc), operation (upsert/insert/update/delete), partition_key, record_key, schema_version, a monotonic sequence, occurred_at/ingested_at, payload (null only for a delete) and provenance (connector id + version + run id + cursor).

Ordering is the contract. Per page:

read/resume → normalize → schema conformance → quality gate
  → ledger PEEK → SINK → ledger COMMIT → checkpoint COMMIT

Both commits happen after the sink resolves. The two tempting shortcuts are both data-loss bugs, and one of them was written here first — see STATUS.md's "A real bug this pass wrote and then caught".

The resulting guarantee is at-least-once, not exactly-once. envelope_id is a deterministic hash of (source, partition, record_key, content) so a dedup-capable sink survives a re-delivery.

P063 — schema evolution

RecordSchema is a small structural shape (field → type + required) plus a CompatibilityPolicy (backward / forward / full / none). checkCompatibility answers what breaks, in which direction, and why. It is deliberately not a second JSON-Schema dialect: the question is "is version N+1 of an UPSTREAM shape safe for consumers of version N?", which needs a comparable structure, not a validator.

Two enforcement points, both upstream of any consumer write:

1. the runner's discovery gate — a shape change that breaks the previous version's declared policy fails the whole run before the first read();
2. per-record conformance — a source that drifted without redeclaring its schema is caught on the offending record, before the page reaches the sink.

A producer cannot widen its own policy: the PREVIOUS version's policy governs whether a change is allowed.

P064 — quality with consequences

QualityGate composes the existing ExpectationEngine (the Great Expectations core subset already in this repo) and adds the missing half: a declared on_violation action.

• block — the page fails, and the checkpoint does not advance, so the bad page stays unconsumed rather than being stepped over.
• quarantine — failing envelopes are persisted whole to opt_quality_quarantine (recoverable, not merely counted); clean ones continue and the checkpoint advances.
• warn — everything continues, violations are counted and reported.

There is no default. With no policy the run report says quality_policy: "none_declared" — an undeclared policy must never read as a passed one. An expectation the local engine cannot execute counts as a VIOLATION, never a silent pass.

P067 — queryable lineage

LineageEmitter keeps emitting OpenLineage to AUD0, and AUD0 remains the system of record. LineageStore adds 0pt's own flattened projection — one row per (run, input, output) with a deterministic edge_id — because "emitted to AUD0" was never the same thing as "answerable from 0pt".

GET /api/0pt/lineage?tenant_id=…&output_name=0pt:dataset:ds_x   # what fed it
GET /api/0pt/lineage?tenant_id=…&input_name=source/src_x        # what it produced

An unfiltered dump is refused (400): that is an export, not a query.

P068 — bounded, idempotent backfill

opt_ingestion_ledger holds one row per (source_id, record_key) with the last accepted content hash. A backfill runs through the same envelope, schema, quality and ledger path as a normal run, and:

• never touches the live checkpoint — a backfill must not rewind or advance streaming ingestion;
• re-emits nothing on a repeat — every record comes back duplicate;
• still re-emits a genuine content change — idempotency is not amnesia;
• refuses an unbounded range at the connector contract level.

Running the real-infrastructure proof

docker run -d --name opt-ingestion-pgtest \
  -e POSTGRES_USER=opt -e POSTGRES_PASSWORD=opt_test_pw -e POSTGRES_DB=opt_test \
  -p 15434:5432 postgres:16-alpine

INGESTION_TEST_DATABASE_URL=postgres://opt:opt_test_pw@localhost:15434/opt_test \
  pnpm vitest run tests/p0/ingestion-plane-postgres.realinfra.test.ts

The suite applies sql/002_source_registry.sql and sql/007_ingestion_plane.sql itself (both are idempotent). It skips honestly, never fabricating success, when the env var is unset.