The Ingestion Pipeline Is an ETL Job

Chunking decided the shape of your rows. Nothing has written any yet. The job that turns 250,000 source documents into roughly 400,000 chunks is a batch job: it reads, transforms, calls a paid API, and writes to a database.

So the questions are ETL questions. Is it idempotent? What happens when it dies two thirds through? Ingestion is the most operationally boring part of RAG and the part that breaks most often, because teams write it as a script and then need it to behave like a pipeline.

Five stages and one gate

The pipeline has five stages, and exactly one of them costs money. Extract pulls the source bytes. Normalize strips what is not content: navigation chrome, boilerplate footers, duplicate whitespace. Chunk applies last lesson's decisions. Embed calls the provider. Upsert writes rows and vectors.

Between chunk and embed sits the content hash gate. Everything before it is cheap, local, and repeatable. Everything after it is billed per token and rate limited.

The content hash is the whole design

Hash the chunk text, store the hash beside the vector, and skip the embed call when the hash has not changed. A SHA-256 of the normalized chunk is enough: on a rerun you read a document's stored hashes in one query, diff them against the fresh ones, and send only the misses. Upsert alone is not enough: if an edit shortens a document, its old high-ordinal rows survive with valid vectors and keep getting retrieved, so delete past the new chunk count in the same transaction.

Those 400,000 chunks at 250 tokens each is 100 million tokens: about $2.00 on text-embedding-3-small at $0.02 per million, or $13 on text-embedding-3-large at $0.13 per million. Rerun it after a week in which 3% of documents changed and the gate turns 100 million tokens into 3 million and six cents. Only the embed stage of the wall clock shrinks; the earlier stages still run over everything. The dollars were never the problem. The hours and the quota were.

-- (document_id, ordinal, embedding_model) is the identity: a rerun updates in
-- place, and two model generations coexist during a swap.
INSERT INTO doc_chunks
  (tenant_id, document_id, ordinal, content, content_hash, embedding_model, embedding, updated_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, now())
ON CONFLICT (document_id, ordinal, embedding_model) DO UPDATE
   SET content      = EXCLUDED.content,
       content_hash = EXCLUDED.content_hash,
       embedding    = EXCLUDED.embedding,
       updated_at   = now()
 WHERE doc_chunks.content_hash <> EXCLUDED.content_hash;
 
-- Last batch, same transaction: drop the tail a shortened document no longer has.
DELETE FROM doc_chunks
 WHERE document_id = $1 AND embedding_model = $2 AND ordinal >= $3;  -- new chunk count
-- tenant_id rides along: lesson 5 filters on it inside the search.

Batch against the documented limits, not a guessed number

Every embedding endpoint publishes two caps per request, an array length and a token total, and the token total usually binds first. OpenAI's create-embeddings endpoint takes 2,048 array entries, 8,192 tokens per input, and 300,000 tokens summed across all inputs. At 250 tokens a chunk that is 1,200 chunks per request, not 2,048. Voyage caps a request at 1,000 texts with a per-model token budget: 120,000 tokens on voyage-4-large, roughly 1,000,000 on voyage-4-lite. Cohere's synchronous embed accepts 96 items per call.

So pack by token budget with array length as a secondary ceiling, and leave slack: sizing to exactly 300,000 tokens means one bad tokenizer estimate turns into a 400. The 429s that arrive anyway are the same token-bucket problem from module 2, so reuse the rate limits lesson's concurrency and backoff rules rather than a second retry policy.

The batch endpoints cover embeddings and are worth pricing: OpenAI's is 50% off inside a fixed 24-hour window, capped at 50,000 inputs; Voyage's is 33% off in 12 hours, 100,000 inputs. Both trade latency for money: fine for a first backfill, bad while you are still debugging your chunker.

It will die at document 250,000

Persist per-document status, not a global cursor. A cursor (offset, last_id) assumes stable ordering and a single worker, and ingestion has neither. Keep a row per document carrying source_updated_at, indexed_at, status, attempts, and last_error. The failure path writes those last two, and the queue query skips rows past three attempts, or a deterministic failure retries forever. That is a work queue, which you have already built.

The other half is transaction size. One transaction around 400,000 rows holds a snapshot open for hours, blocks autovacuum, and loses everything on a single connection reset. Bound each transaction to the batch you just embedded, a few hundred to a few thousand rows, the same bulk write sizing as any large import.

async function ingestDocument(doc: Doc) {
  const chunks = chunk(normalize(doc.body)).map((text, ordinal) => ({
    text, ordinal, hash: sha256(text),
  }));
  const known = await db.hashesFor(doc.id, MODEL);      // ordinal -> hash, one query
  const stale = chunks.filter((c) => known.get(c.ordinal) !== c.hash);
  metrics.embedSkipped.inc(chunks.length - stale.length);
 
  // 250k tokens, 1k entries: the tightest caps across providers.
  for (const batch of packByTokens(stale, 250_000, 1_000)) {
    const vectors = await embed(batch.map((c) => c.text));   // 429s handled upstream
    await db.tx(async (t) => {                               // one tx per batch
      await upsertChunks(t, doc.id, MODEL, batch, vectors);  // the ON CONFLICT above
      await t.touchDocument(doc.id, "in_progress");
    });
  }
  await db.pruneBeyond(doc.id, MODEL, chunks.length);       // the DELETE above
  await db.markIndexed(doc.id, doc.sourceUpdatedAt);
}

A backfill is a migration, not a script

A first full ingest runs for hours against the same database and provider quota your live API depends on, so treat it as a migration. Four requirements follow.

• Its own rate limit, set below the provider ceiling. A backfill that saturates your embeddings quota takes the request path down with it.
• Its own database budget. Read from a replica where you can, keep writes batched, and expect autovacuum to fall behind on the chunks table.
• Index creation last. Load the vectors first, then build the index with CREATE INDEX CONCURRENTLY, which pgvector's README recommends for production. Raising maintenance_work_mem so the graph fits in memory is the difference between minutes and hours.
• An owner watching it. Multi-hour jobs with no dashboard get killed by a routine deploy, and nobody finds out for a week.

The counters that tell you the index is stale

Seven numbers make ingestion observable, and only the last one is visible to users. Documents seen, chunks written, embed calls made, chunks skipped by the gate, tokens spent, dollars spent, indexing lag.

The skip counter is your gate's hit rate and a canary: a rerun that suddenly skips nothing means a normalizer change invalidated every hash, and a full re-embed you did not intend.

Indexing lag is the metric that matters. Define it as now() minus the source_updated_at of the oldest document whose status is not indexed, and alarm on it. It is the replication lag of your retrieval system: when it climbs, the model answers from a stale copy of your data, confidently, with no error in your logs. Keeping it low once rows change one at a time needs change capture rather than a batch job, which is its own guide.

Key takeaways

• Ingestion is an ETL job with five stages (extract, normalize, chunk, embed, upsert) where only embed costs money, so the design is about avoiding it.
• A content hash gate between chunk and embed cuts the paid stage: 3% churn means embedding 3% of the chunks. Idempotency also needs a delete past the new chunk count.
• Key the gate on (content_hash, embedding_model) and hash the post-normalization string, or a whitespace fix invalidates everything and a model swap invalidates nothing.
• Batch by token budget, not item count: OpenAI allows 2,048 inputs but only 300,000 tokens per request, Voyage 1,000 texts, Cohere 96.
• Resume from per-document status rows, never a global offset, and bound each transaction to one embedded batch instead of the whole corpus.
• Indexing lag, the age of the oldest document not yet indexed, is the one ingestion metric users can feel: stale retrieval produces confident wrong answers and zero errors.