Ingestion Design

The ingestion layer extracts raw data from OEM-specific external APIs, validates it, and progressively refines it into consolidated entity records partitioned by day. Each OEM runs its own independent Dagster deployment. Shared infrastructure lives in a common library so OEM pipelines don't share state or affect each other.

Pipeline Structure

Every ingestion pipeline follows the same three-stage structure:

Stage	Responsibility
Raw	Fetch data from the source API and store the response exactly as returned
Transformed	Extract entity records from the raw response, deduplicate by key group, and validate
Consolidated	Merge transformed records across all sources for an entity type

Each stage has asset checks that run automatically after materialization.

Dependency Ordering

The asset DAG has two independent ordering dimensions that do not form cycles:

• Source dependencies operate in the raw and transformed tiers. One raw source may depend on another (e.g. fetching search results requires dealer IDs from a prior fetch). These dependencies are fully resolved by the end of the transformed tier.

• Entity dependencies begin at the consolidated tier, where foreign_keys allow one entity's consolidation to reference another consolidated entity's same-partition data. For example, consolidated inventory can reference consolidated models to attach a resolved model ID to each row. Since all source-level dependencies are already resolved, entity ordering is independent of source ordering.

This separation guarantees no circular dependencies: sources never depend on entities, and entities never depend on sources.

Asset Key Conventions

Dagster asset keys form a path hierarchy using / as the separator:

Stage	Pattern	Example
Source	<oem>/<market>/sources/<source>_<resource>	audi/us/sources/scs_search
Raw	<oem>/<market>/raw/<source>_<resource>	audi/us/raw/scs_search
Transformed	<oem>/<market>/transformed/<name>_<entity>	audi/us/transformed/onegraph_inventory
Consolidated	<oem>/<market>/consolidated/<entity>	audi/us/consolidated/inventory

Asset Checks

Stage	Check name	What it verifies	Blocking
Raw	fetch_attempts	All HTTP requests made by the fetcher returned 2xx; per-status failure patterns recorded in metadata	No
Transformed	discrepancies	No conflicting rows for the same entity ID; same-source field disagreements grouped into discrepancy patterns	No
Consolidated	integrity	Every row has a non-null entity ID; records row count	No
Consolidated	unresolved_refs	Cross-entity reference links resolved successfully	No
Transformed (highest-priority source)	cross_source_discrepancies	Cross-source merges produce no conflicting field values for the same entity; field-level disagreements grouped into discrepancy patterns	No
Transformed (paired inventory sources)	cross_source_overlap	Membership overlap between two inventory feeds stays within an expected range; per-side coverage and a breakdown of keys missing from the other feed recorded in metadata	Configurable

Most checks are non-blocking (severity WARN) — they surface issues in the Dagster UI without aborting the run. The fetch_attempts check additionally reports tracking=instrumented or tracking=uninstrumented in its metadata so operators can tell whether the source module is recording per-request outcomes (see Component Reference).

The one exception is cross_source_overlap: when its blocking flag is set it emits at severity ERROR and does short-circuit the run. See ConsolidatedComponent → Cross-source overlap for configuration and the run-topology caveat.

Failing checks are triaged per the rules in Verification → Alert triage — potential raw-collection issues take priority over everything else, because raw gaps are permanent.

Response Integrity

A truncated response can look complete at the status line: the request succeeded, the status is 200, and only the payload is short. Left unchecked it is written to the raw tier as a partial JSON body and fails days later as a JSONDecodeError inside an OEM extractor.

Truncation splits into two cases, handled at two different layers.

Framing truncation — the transport's job. A body short of its declared Content-Length never becomes a response object: httpx raises RemoteProtocolError, libcurl raises CURLE_PARTIAL_FILE (18) or CURLE_RECV_ERROR (56). default_request_error_handler retries all three. There is deliberately no Content-Length check in application code — it would be unreachable. packages/ai_http/tests/test_errors.py pins that with a socket server that lies about the header, asserting both transports raise before anything can inspect a body.

Payload truncation — what the transport cannot see. An origin or CDN that truncates the JSON upstream and then frames the partial bytes correctly produces a complete, valid HTTP message whose payload is short. Content-Length matches the truncated body, so only the payload's own structure gives it away. Every successful response therefore passes raise_for_incomplete_body inside the request machinery itself — the shared HttpSession.request_events loop (every session type, every paginator) and the standalone http_get_with_retry / http_post_with_retry helpers. The check is a single comparison, independent of body size, so a 100 MB envelope costs the same as a 100 byte one. A body whose first significant character is { or [ must end with the matching } or ], examining only a bounded window at each end.

It runs before _validate_and_get_bytes, so a cut payload is reported as truncation rather than as whatever the validation hook trips over first. Response validators that parse the body (GM's AEC check, Genesis's dealer envelope check) otherwise fail with an opaque ijson error. IncompleteJSONError is not a ValueError, so their own guards do not catch it.

A body that fails the check raises TruncatedResponseError, a subclass of TransientHTTPError, so the response is a failed request: retried on the transport's own budget, and on exhaustion recorded as a failed attempt in the raw asset's fetch_attempts check. Custom RequestErrorHandlers must therefore branch on the exception type, not on the status code — a truncated body carries the 2xx it arrived with, so a status-only handler would abort the one failure a retry reliably fixes.

Only 2xx bodies are checked. On a failure status the status itself is the finding, and a cut error page must not mask it — nor spend the retry budget on a status that would not be retried on its own.

Two limits are worth knowing:

• The terminator check is a necessary, not sufficient, condition. A stream cut exactly on an inner object boundary ({"items":[{…},{…}) still terminates plausibly. Making it exact means a full parse, which allocates several times the body — the memory peak the guard exists to avoid. The check fails open for the same reason: wrongly storing a suspect body costs one row, wrongly rejecting a good one drops a page permanently.
• One rejected page does not reject the partition. The truncated document is never stored, but with abort_on_failure=False (the default for the paginated helpers) the surviving pages are still written, so the partition can be short by one page. fetch_attempts is what flags it — a successful materialization with a failed fetch_attempts check means exactly this.

Source modules that call httpx directly rather than going through ai_http bypass the guard entirely. New sources should use the session helpers.

Asset Groups and Tags

Source observability assets are grouped sources. Raw fetch assets are grouped raw. Transformed assets are grouped transformed. Consolidated assets are grouped consolidated. Groups are visible in the Dagster UI's asset catalog and can be used to filter and select assets in bulk.

Every ingestion asset also carries an oem tag for per-OEM filtering:

oem: audi

Partitioning

All ingestion assets use a DailyPartitionsDefinition with end_offset=1. The end_offset=1 means today's date is always a valid partition, so assets can be materialized on the day they are due without waiting for the day to roll over.

Request Throttling

Fan-out HTTP fetches are paced by ai_http.Throttle, a pool-wide request pacer shared across every worker thread and retry in a fan-out. It enforces steady-state spacing (min_interval seconds between requests across the pool) and a shared 429 penalty window (penalize() pauses the whole pool, not just the worker that hit the limit).

By default the throttle keeps its state in-process, so it only coordinates threads within a single Dagster run. When the same origin (or a shared API budget) is hit by multiple concurrent runs — backfills, overlapping partitions, or several sources behind one rate limit — a per-run throttle cannot see the aggregate load. Throttle therefore accepts a pluggable backend: the default LocalBackend (in-process) or a DynamoThrottleBackend that stores the pacing state in a shared DynamoDB item, so every process sharing a key forms one pool.

Distributed throttle

Construct a cross-process throttle with ai_http.distributed_throttle(key, min_interval, *, local_min_interval=None, meter=None). The key is a stable string identifying the shared budget:

• Per-credential limits → one key per credential. Zyte's rate limit is per API key, so zyte_throttle(prefix) builds the key from a prefix (e.g. "BMW_STOLO" → "zyte/bmw_stolo"). ZyteSession/ZyteSessionFactory take that same key_prefix and use it for both the throttle and the <PREFIX>_ZYTE_API_KEY credential lookup, so a call site's pacing and credential stay paired. Call sites sharing a prefix share one pool; distinct keys are paced independently. Only the prefix is used — the secret value never reaches the throttle store.
• Per-origin limits where only concurrent runs of the same source contend → "<oem>/<source>", e.g. "gm/inventory".

Coordination is a single conditional UpdateItem (compare-and-set) per granted slot — no lock is held, so a crashed process cannot block the pool. Timestamps are wall-clock epoch seconds. NTP skew between tasks is far below any min_interval in use, and the CAS is on stored state, so skew can only blur spacing, never corrupt it. Items carry an expires_at TTL so idle keys clean themselves up.

Enabling it. The backend is only used when DISTRIBUTED_THROTTLE_TABLE is set (to the DynamoDB table name, distributed-throttle in prod). Unset, distributed_throttle() returns a plain in-process Throttle at local_min_interval. This env var is the per-project kill-switch — local dev, unit tests, and un-opted-in deployments are safe by default. DISTRIBUTED_THROTTLE_REGION (optional — unset lets boto3 resolve the region normally, e.g. from AWS_REGION or task metadata) and DYNAMODB_LOCAL_ENDPOINT (local override) are also honored. The run task's IAM role needs dynamodb:UpdateItem and dynamodb:GetItem on the table.

Contention vs. outage. Every grant and every failed compare-and-set is a write to one hot partition key, so a large fan-out can make DynamoDB throttle the writes. That is contention, not an outage. The backend treats a throttling response as a back-off-and-retry signal (exponential backoff keyed on DynamoDB's own throttling), so the retries spread out until the writes succeed, rather than counting it toward the outage fallback.

Fail-open. DynamoDB availability is not a correctness dependency: on genuine connectivity errors (three consecutive non-throttling, non-conditional failures) the backend logs once and paces in-process at max(min_interval, local_min_interval) for a bounded window, then re-probes DynamoDB. A DynamoDB outage costs cross-run coordination for that window but never a run.

Metering. An injected RequestMeter records this process's share of the offered load, not the pool-wide aggregate. Read each run's meter summary to reconstruct per-process contribution.