Chapter 24 · Ingestion

Pulling from a document store

The first adapter. Pagination that loses documents, permission metadata that arrives in three shapes, and why the modified timestamp you were given is not a change signal.

12 min read

The first concrete adapter, implementing the discover and fetch stages from the pipeline for a document store. Atlas has three document sources — policies, specifications and contracts — and they all come through this adapter with different configuration.

Document stores differ enormously in their APIs and are almost identical in the three ways they cause problems.

The adapter interface

// src/ingest/sources/types.ts — every source implements this, and nothing
// outside the adapter knows which vendor it is talking to.
export interface SourceAdapter {
  /** An async generator, not an array. A store with 50,000 documents must not
   *  be materialised in memory before the first one is processed. */
  discover(): AsyncIterable<DiscoveredDocument>
  fetch(doc: DiscoveredDocument): Promise<FetchedDocument>
}

Returning an AsyncIterable from discover rather than a Promise<DiscoveredDocument[]> is a small decision with two large consequences: memory stays flat regardless of corpus size, and the first document starts parsing while page two is still being fetched.

Problem 1: pagination that loses documents

Almost every document store paginates with offset and limit, and almost every one of them is unstable under concurrent writes.

Reading a document twice is harmless — the ingest is idempotent. Never reading one is the problem, and it is invisible: the run reports 2,399 documents seen, which looks exactly like 2,400.

Three fixes, in order of preference.

// ✅ Best: a cursor the server defines. Stable across writes.
async function* discoverByCursor(): AsyncIterable<DiscoveredDocument> {
  let cursor: string | undefined
  do {
    const page = await api.list({ cursor, limit: 100 })
    for (const item of page.items) yield toDiscovered(item)
    cursor = page.nextCursor
  } while (cursor)
}

// ✅ Good: keyset pagination on a stable, totally ordered key.
async function* discoverByKeyset(): AsyncIterable<DiscoveredDocument> {
  let after = ''
  for (;;) {
    const items = await api.list({ orderBy: 'id', after, limit: 100 })
    if (items.length === 0) return
    for (const item of items) yield toDiscovered(item)
    after = items[items.length - 1].id
  }
}

// ⚠️ Last resort: offset pagination, with a reconciliation pass.
async function* discoverByOffset(): AsyncIterable<DiscoveredDocument> {
  const seen = new Set<string>()
  for (let offset = 0; ; offset += 100) {
    const items = await api.list({ offset, limit: 100 })
    if (items.length === 0) break
    for (const item of items) {
      if (seen.has(item.id)) continue
      seen.add(item.id)
      yield toDiscovered(item)
    }
  }
  // The count check from chapter 23, made specific: compare against the
  // store's own total. A mismatch means the pagination dropped something.
  const total = await api.count()
  if (seen.size < total) {
    throw new IngestAnomaly(`discovered ${seen.size} of ${total} documents`)
  }
}

That final comparison is the part worth copying even when the pagination is sound. It costs one request and it converts a silent gap into a failed run.

Problem 2: the modified timestamp lies

Every store has a modifiedAt, and it means something different in each one. Before using it as a change signal, find out which of these it is:

What it can mean	Consequence for sync
The content changed	✅ What you want
Any metadata changed — a tag, a move, a share	Re-ingests unchanged content. Wasteful, harmless
The record was touched by a migration	Re-ingests everything, once, expensively
It is set on write and never updated	Silently misses every edit

The last row is the dangerous one, and it is more common than it should be — particularly in stores that treat an edit as a new version rather than a mutation.

// Do not trust it. Combine with a content hash, which is authoritative.
export async function hasChanged(doc: DiscoveredDocument): Promise<boolean> {
  const known = await db.query(
    `select content_hash, source_updated_at from documents where id = $1`,
    [documentId(doc)],
  )
  if (known.rowCount === 0) return true

  // Cheap signals first: if BOTH say unchanged, skip without fetching.
  const row = known.rows[0]
  if (doc.etag && doc.etag === row.etag) return false
  if (+doc.updatedAt <= +row.source_updated_at && !FORCE) return false

  // Otherwise fetch and hash. Fetching is cheap; embedding is not, and the
  // hash is what stops an unchanged document from being re-embedded.
  return true
}

The design that follows from this is in chapter 34: timestamps decide whether to fetch, and content hashes decide whether to embed. Getting those two gates the right way round is most of what makes incremental sync work.

Problem 3: permission metadata in three shapes

Chapter 14 established that classification is derived from source metadata, never from parsed text. Document stores expose that metadata in one of three shapes, and the adapter normalises all three.

// Shape A: a folder path. The most common, and the most fragile.
function classifyByPath(path: string): Classification {
  // ⚠️ An allow-list, as in chapter 16. A path that matches nothing is a skip,
  // not a default — a new folder must not inherit the loosest classification.
  if (path.startsWith('policy/')) return 'public_internal'
  if (path.startsWith('specs/')) return 'public_internal'
  if (path.startsWith('contracts/')) return 'account'
  if (path.startsWith('finance/')) return 'financial'
  throw new SkipDocument('transform', `unclassified path: ${path}`)
}

// Shape B: custom fields on the document record. The best case.
function classifyByFields(meta: Record<string, unknown>): Classification {
  const value = meta.confidentiality
  if (typeof value !== 'string' || !isClassification(value)) {
    throw new SkipDocument('transform', `missing or invalid confidentiality field`)
  }
  return value
}

// Shape C: an ACL per document. Chapter 13's third model.
async function classifyByAcl(doc: FetchedDocument) {
  const grants = await api.listPermissions(doc.sourceRef)
  return {
    classification: 'acl' as const,
    principals: grants.map((g) => g.principalId),
  }
}

Throwing SkipDocument on an unclassifiable document, rather than defaulting, is the rule this chapter would most want you to take away. A default classification is wrong in one of two directions and the loose direction is a leak.

The Atlas adapter

// src/ingest/sources/document-store.ts
export function documentStoreAdapter(cfg: DocumentStoreConfig): SourceAdapter {
  return {
    async *discover() {
      let cursor: string | undefined
      do {
        const page = await withRetry(() =>
          cfg.client.list({ folder: cfg.folder, cursor, limit: 100 }),
        )
        for (const item of page.items) {
          // Filter before yielding: a document excluded here never costs a
          // fetch, a parse or an embedding.
          if (cfg.exclude?.(item)) continue
          yield {
            sourceId: cfg.sourceId,
            sourceRef: item.id,
            title: item.name,
            updatedAt: new Date(item.modifiedAt),
            etag: item.etag,
            size: item.size,
          }
        }
        cursor = page.nextCursor
      } while (cursor)
    },

    async fetch(doc) {
      const [bytes, meta] = await Promise.all([
        withRetry(() => cfg.client.download(doc.sourceRef)),
        withRetry(() => cfg.client.metadata(doc.sourceRef)),
      ])

      // A zero-byte or truncated download is a skip, not a document. Parsing
      // one produces an empty block list, which indexes a document with no
      // content — findable, cited, and empty.
      if (bytes.length === 0) {
        throw new SkipDocument('fetch', `empty download for ${doc.sourceRef}`)
      }
      if (doc.size && bytes.length < doc.size * 0.9) {
        throw new SkipDocument('fetch', `truncated: ${bytes.length} of ${doc.size}`)
      }

      return { ...doc, bytes, contentType: meta.contentType, sourceMetadata: meta.fields }
    },
  }
}

The truncation check exists because of how this fails without it. A partial download parses successfully, chunks successfully, embeds successfully, and produces a document whose last two thirds are missing. Retrieval finds the first third happily and nothing indicates the rest existed.

Retries that distinguish the retryable

// Not every error should be retried, and retrying the wrong ones turns a
// permission problem into a rate-limit problem.
const RETRYABLE = new Set([408, 429, 500, 502, 503, 504])

export async function withRetry<T>(fn: () => Promise<T>, attempts = 4): Promise<T> {
  for (let i = 0; ; i++) {
    try {
      return await fn()
    } catch (error) {
      const status = (error as { status?: number }).status
      // 403 means you are not allowed. Four more attempts will not help, and
      // they look like an attack to whoever reads the store's audit log.
      if (!status || !RETRYABLE.has(status) || i >= attempts - 1) throw error

      const retryAfter = Number((error as { headers?: Record<string, string> })
        .headers?.['retry-after'])
      const delay = Number.isFinite(retryAfter)
        ? retryAfter * 1000
        : Math.min(2 ** i * 500 + Math.random() * 500, 30_000)

      logger.warn('fetch.retry', { attempt: i + 1, status, delayMs: Math.round(delay) })
      await sleep(delay)
    }
  }
}

Honouring Retry-After when the server sends it is not politeness. A fixed exponential backoff against a server that has told you to wait 60 seconds produces five failed attempts and then a failure, where one correct wait would have succeeded.

The failure mode: the folder that was renamed

Atlas's contracts live under contracts/. Someone reorganises the store and moves active contracts to legal/contracts/active/, leaving the old folder in place with the pre-2023 archive.

The next run discovers 400 documents instead of 3,100. The anomaly check from chapter 23 catches that — a drop to 13% of the previous count fails the run loudly, which is the system working.

The subtler version is the one to watch for. Suppose instead that specs/ gains a subfolder specs/drafts/, holding working versions of specifications that have not been approved. Nothing is removed, the count goes up by 300, no threshold fires, and 300 draft specifications are now indexed as authoritative product data with the same classification as the approved ones.

// The check for growth, which is the direction nobody alerts on.
if (previous > 0 && result.seen > previous * 1.2) {
  logger.warn('ingest.unexpected_growth', {
    source: source.id, seen: result.seen, previous,
    newPaths: [...newTopLevelPaths].slice(0, 10),
  })
}

A warning rather than a failure, because growth is usually legitimate — but a warning that names the new paths, so the person reading it can tell a bulk upload from a folder that should have been excluded. ⚠️ Pair it with the allow-list from problem 3: with path-based classification throwing on unmatched paths, specs/drafts/ would have been skipped rather than silently absorbed, and the skip report would have named it.

That is the pattern worth internalising from this chapter: the count checks tell you something changed, and the allow-list decides what happens by default. Neither is sufficient alone.

Key points

discover() returns an async iterable, so memory stays flat and parsing overlaps fetching.
Offset pagination drops documents under concurrent writes. Prefer a cursor or keyset; if you cannot, reconcile against the store's own count.
Find out what the store's modifiedAt actually means. The worst case — set on write, never updated — silently misses every edit.
Timestamps gate fetching; content hashes gate embedding.
Normalise permission metadata in the adapter, and skip unclassifiable documents rather than defaulting them.
Check for empty and truncated downloads. A partial document indexes successfully with two thirds missing.
Retry only retryable statuses, and honour Retry-After.
Alert on counts dropping and growing. Growth is the direction nobody watches, and it is how unapproved drafts get indexed as authoritative.

Next: pulling from a relational database — the free-text columns that belong in the index, and keeping a consistent read while the table moves underneath you.