190 lines
8.9 KiB
Markdown
190 lines
8.9 KiB
Markdown
# 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
|
||
|
||
```bash
|
||
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.
|