Chapter 23 · Ingestion

The ingestion pipeline, end to end

Six stages, the contract between them, and the two properties that make a pipeline debuggable: every stage is resumable and every document can fail alone.

12 min read

Section 3 builds the offline half of the architecture. It is the longest section in the series, and almost every chapter in it is one stage of the pipeline this chapter lays out.

It assumes the registry, which already knows every source's shape, route, classification and schedule, and the four findings from the dry run — superseded versions, unlinked amendments, the inference path, and inconsistent titles. Three of those four are implemented in this section.

The six stages

Each has a different failure mode, a different cost, and a different retry policy, and keeping them separate is what makes a bad answer traceable back to a stage rather than to "ingestion".

Stage	Fails because	Cost	Retry
Discover	Source unreachable, auth expired	Trivial	Whole run
Fetch	Network, rate limit, permission	Low	Per document, with backoff
Parse	Corrupt file, no text layer, odd format	Medium	Never — it will fail again
Transform	Unresolvable reference, missing parent	Trivial	Never
Chunk	Pathological structure	Trivial	Never
Embed & store	Provider error, dimension mismatch	Highest	Per batch, with backoff

The cost column drives the design. Embedding is the only stage that costs real money per document, so it is the last stage and everything cheap happens before it. A document rejected at parse has cost nothing.

The contract between stages

One type per boundary, and each stage is a pure function of the previous one wherever it can be.

// src/ingest/types.ts — the pipeline's spine. Each stage consumes the type
// above it and produces the one below.

/** 1. Discover: identity and a change signal. No content. */
export interface DiscoveredDocument {
  sourceId: string
  sourceRef: string           // the id in the system of record
  title: string
  updatedAt: Date
  /** Whatever the source gives you to detect change cheaply. */
  etag?: string
  size?: number
}

/** 2. Fetch: the bytes, plus the metadata the source system holds. */
export interface FetchedDocument extends DiscoveredDocument {
  bytes: Buffer
  contentType: string
  /** ⚠️ Classification is derived from THIS, never from parsed text. Chapter 14. */
  sourceMetadata: Record<string, unknown>
}

/** 3. Parse: structured blocks, not a string. Structure survives to chunking. */
export interface ParsedDocument extends Omit<FetchedDocument, 'bytes'> {
  blocks: Block[]
  parseWarnings: string[]
}

export type Block =
  | { type: 'heading'; depth: number; text: string }
  | { type: 'paragraph'; text: string }
  | { type: 'table'; caption?: string; headers: string[]; rows: string[][] }
  | { type: 'list'; ordered: boolean; items: string[] }
  | { type: 'code'; language?: string; text: string }

Parsing to blocks rather than to a string is the decision that matters most in this file. A string has already lost the information that chapter 32 needs — which text was a heading, where a table began — and no amount of regex recovers it reliably. Every parser in chapters 24 to 29 produces blocks.

/** 4. Transform: the dry run's findings applied. Still whole documents. */
export interface TransformedDocument extends ParsedDocument {
  classification: Classification
  accountId: string | null
  supersedes: string[]        // ids this version replaces
}

/** 5. Chunk. 6. Embed. */
export interface PreparedChunk {
  ordinal: number
  content: string             // what the model sees
  embedText: string           // what gets embedded — not always the same
  headingPath: string[]
  kind: 'prose' | 'table' | 'list' | 'code'
  tokens: number
}

content and embedText as separate fields is the pattern from chapter 9, promoted into the type so it cannot be forgotten. Prefixing the heading path onto the embedded text measurably improves retrieval on fragments; prefixing it onto the displayed text makes answers read oddly.

Two properties that make it debuggable

Every document can fail alone

A 2,400-document ingest that aborts on document 1,900 because one PDF is corrupt is an ingest you will run four times. The control flow is per-document, and a failure is recorded rather than thrown.

// src/ingest/run.ts
export class SkipDocument extends Error {
  constructor(readonly stage: Stage, message: string) { super(message) }
}

export async function ingestSource(source: SourceDefinition, adapter: SourceAdapter) {
  const result: SyncResult = { seen: 0, changed: 0, skipped: 0, failed: 0, ok: true }
  const skips: Array<{ ref: string; stage: Stage; reason: string }> = []

  for await (const discovered of adapter.discover()) {
    result.seen++
    try {
      if (!(await hasChanged(discovered))) continue    // chapter 34
      await ingestOne(source, adapter, discovered)
      result.changed++
    } catch (error) {
      if (error instanceof SkipDocument) {
        // Expected: a scan with no text layer, a contract with no account.
        result.skipped++
        skips.push({ ref: discovered.sourceRef, stage: error.stage, reason: error.message })
        continue
      }
      // Unexpected: record and keep going, but the run is not clean.
      result.failed++
      result.ok = false
      logger.error('ingest.failed', { ref: discovered.sourceRef, error: String(error) })
    }
  }

  // ⚠️ Skips are reported, never silent. A source that skipped 760 of 1,900
  // documents "succeeded" by every other measure. Chapter 36 reads these.
  await writeSkipReport(source.id, skips)
  return result
}

The distinction between skipped and failed is worth keeping. A skip is a document you decided not to index for a stated reason; a failure is something you did not anticipate. A run with 760 skips and 0 failures is healthy if you expected 760 scans. The same run with 0 skips and 760 failures is broken.

Every stage is resumable

Embedding 45,000 chunks takes a while and will be interrupted. Resumability comes from making the unit of work idempotent, not from a checkpoint file.

// One document, one transaction. Re-running it is safe and produces the same
// result, which means "resume" is just "run it again".
async function ingestOne(source, adapter, discovered: DiscoveredDocument) {
  const fetched = await adapter.fetch(discovered)
  const parsed = await parse(fetched)                       // chapters 24-29
  const transformed = await transform(parsed, source)       // chapter 14, 22
  const chunks = chunkDocument(transformed)                 // chapters 30-32
  const vectors = await embedBatched(chunks.map((c) => c.embedText))

  await db.transaction(async (tx) => {
    await upsertDocument(tx, transformed)
    // Replace, never append. Re-ingesting must not leave stale chunks —
    // which is half of chapter 35's problem.
    await tx.query(`delete from chunks where document_id = $1`, [transformed.id])
    await insertChunks(tx, transformed.id, chunks, vectors)
  })
}

The transaction boundary is the whole document. A partially chunked document in the index is worse than an absent one: retrieval returns half a policy and the answer is confidently incomplete.

Where the time and money go

Measured on Atlas's first full run, 2,400 documents and roughly 45,000 chunks:

Stage	Share of wall time	Share of cost
Discover	<1%	0
Fetch	18%	0
Parse	31%	0 (OCR excluded)
Transform	<1%	0
Chunk	2%	0
Embed	48%	~100%

Two consequences for how you work.

Cache the parse output during development. You will re-run the pipeline dozens of times while tuning chunking, and 31% of each run is re-parsing files that did not change. A content-hashed parse cache turns a twelve-minute iteration into a four-minute one.

// Parse is deterministic and expensive. Hash the bytes, cache the blocks.
const key = createHash('sha256').update(fetched.bytes).digest('hex')
const cached = await parseCache.get(key)
if (cached) return cached

Embedding is effectively the whole bill, so chunk count is the cost lever. Chapter 31 measures chunk size against quality; it is also measuring it against money.

Concurrency, and the limit that matters

// Documents in parallel, with a bound. The right bound is set by the slowest
// shared resource, which is almost never your CPU.
import pLimit from 'p-limit'

const limit = pLimit(8)   // ⚠️ bounded by the SOURCE's rate limit, not yours

await Promise.all(
  documents.map((doc) => limit(() => ingestOne(source, adapter, doc))),
)

Eight is a starting point. The real constraint is whichever of these is tightest: the source system's rate limit, the embedding provider's requests-per-minute, or your database's connection pool. Setting concurrency to 50 against a document store that allows 20 requests per second produces a run that is slower than 8, because every third request is a retry.

// Batch at the embedding boundary rather than raising concurrency. One request
// with 96 inputs beats 96 requests, by a wide margin.
export async function embedBatched(texts: string[], batchSize = 96): Promise<number[][]> {
  const out: number[][] = []
  for (let i = 0; i < texts.length; i += batchSize) {
    out.push(...(await embed(texts.slice(i, i + batchSize))))
  }
  return out
}

The failure mode: the run that reports success

This is the one to design against from the first version, because it is silent and it survives every test.

The pipeline runs nightly. It reports ok: true. Retrieval works. Three weeks later somebody notices that no product specification published since the start of the month is findable.

What happened: the specification source's discover() started returning an empty list, because an auth token expired and the API returns 200 with { "items": [] } rather than 401. Zero documents seen, zero changed, zero failed. Every metric is green because every metric counts things that went wrong, and nothing went wrong — nothing happened at all.

// The check that catches it. An expectation per source, from the registry's
// measured change rate, asserted after every run.
export function assertPlausible(source: SourceDefinition, result: SyncResult) {
  // A source that has documents must discover documents.
  if (result.seen === 0) {
    throw new IngestAnomaly(`${source.id}: discovered 0 documents. Expected > 0.`)
  }
  // A drop of more than a fifth against the last run is a structural change,
  // not a quiet Tuesday.
  const previous = lastRun(source.id)?.seen ?? 0
  if (previous > 0 && result.seen < previous * 0.8) {
    throw new IngestAnomaly(
      `${source.id}: discovered ${result.seen}, down from ${previous}. ` +
      `Check auth and filters before trusting this run.`,
    )
  }
}

Count what did not happen. Every monitoring system counts errors; the expensive failures in an ingestion pipeline are absences.

That is the same principle as chapter 19's sync heartbeat, applied one level down: there, you distinguish "nothing changed" from "nothing ran"; here, you distinguish "nothing changed" from "nothing was found".

Key points

Six stages with one type per boundary. Keeping them separate is what makes a bad answer traceable to a stage.
Parse to structured blocks, never to a string. A string has already lost what chunking needs.
Embedding is the only expensive stage, so it is last and everything cheap filters before it.
content and embedText are separate fields in the type, so the distinction cannot be forgotten.
Every document fails alone, and skips are distinguished from failures and always reported.
One transaction per document, replacing chunks rather than appending. That makes resume mean "run it again".
Cache parse output by content hash during development. It is a third of every run.
Bound concurrency by the source's limit, and batch at the embedding boundary instead.
Assert that a run discovered a plausible number of documents. An empty discovery reports perfect health.

Next: pulling from a document store — the first adapter, and the three things every document store gets wrong.