Chapter 34 · Ingestion

Incremental sync and change detection

Two gates, in the right order: timestamps decide whether to fetch, hashes decide whether to embed. Plus the watermark discipline that makes a missed window recoverable.

11 min read

A full ingest of Atlas's corpus is 2,400 documents, 45,000 chunks, and about an hour. Running that nightly is wasteful; running it hourly for contracts is impossible.

Incremental sync is how the nightly run touches the forty documents that changed. The whole design is two gates in the right order, and the ordering is the part that is usually wrong.

The two gates

Gate 1 is cheap and unreliable. Gate 2 is authoritative and requires a fetch. Running only gate 1 re-embeds documents that were merely touched; running only gate 2 fetches and parses the entire corpus every night.

// src/ingest/sync.ts
export async function syncDecision(
  discovered: DiscoveredDocument,
  known: KnownDocument | null,
): Promise<'skip' | 'fetch'> {
  if (!known) return 'fetch'                      // new
  if (known.deletedAt) return 'fetch'             // resurrected

  // An etag is the source's own content signal and is authoritative when present.
  if (discovered.etag && known.etag) {
    return discovered.etag === known.etag ? 'skip' : 'fetch'
  }

  // Otherwise the timestamp, with the rewind from chapter 25.
  if (+discovered.updatedAt <= +known.sourceUpdatedAt) return 'skip'
  return 'fetch'
}

export async function embedDecision(
  parsed: ParsedDocument,
  known: KnownDocument,
  source: SourceDefinition,
): Promise<'unchanged' | 'metadata_only' | 'reembed'> {
  const hash = hashBlocks(parsed.blocks)
  if (hash !== known.contentHash) return 'reembed'

  // ⚠️ The case the hash cannot see: content identical, classification
  // changed. An account reassignment, a reclassification, a folder move.
  // Missing this leaves a document permission-scoped to its previous owner.
  const classification = resolveClassification(parsed, source)
  if (classification.classification !== known.classification ||
      classification.accountId !== known.accountId) {
    return 'metadata_only'
  }

  return 'unchanged'
}

The metadata_only branch is the one that is usually absent, and its absence is a permission bug rather than a freshness one. A contract that moves from account 4471 to account 4480 has identical content and a different owner, and a hash-only gate decides nothing changed.

// metadata_only is cheap and must not be skipped.
await db.query(
  `update documents
      set classification = $2, account_id = $3, indexed_at = now()
    where id = $1`,
  [known.id, classification.classification, classification.accountId],
)

The watermark

A sync run covers a window. The watermark is where the next window starts, and the discipline around it is what makes a missed run recoverable rather than a permanent gap.

interface Watermark {
  sourceId: string
  /** The point up to which this source is known complete. */
  syncedThrough: Date
  lastRunAt: Date
  lastRunOk: boolean
}

export async function runSync(source: SourceDefinition) {
  const watermark = await getWatermark(source.id)

  // Rewind, per chapter 25. Re-reading five minutes is free; a row hidden by
  // clock skew is invisible forever.
  const from = new Date(+watermark.syncedThrough - 5 * 60_000)

  // ⚠️ Capture the window's end BEFORE the run, not after. Using now() at the
  // end advances the watermark past documents modified during the run, which
  // would then never be seen by any run.
  const windowEnd = new Date()

  const result = await ingestSince(source, from)

  // Advance only on a clean run. A run that failed halfway leaves the
  // watermark where it was, and the next run re-covers the whole window.
  if (result.ok) {
    await setWatermark(source.id, { syncedThrough: windowEnd, lastRunOk: true })
  } else {
    await setWatermark(source.id, { ...watermark, lastRunAt: new Date(), lastRunOk: false })
    logger.error('sync.incomplete', { source: source.id, failed: result.failed })
  }
}

Two rules, and both have the same justification:

Capture the window end before the work. Advance the watermark only past work that is durably done.

That is the same invariant as chapter 26's backfill cursor, and it has the same consequence: re-doing idempotent work is free, skipping work is silent.

Three sources of change, three detection methods

Source	Method	Caveat
Document store	`modifiedAt` or etag + hash	What the timestamp means varies
Database	`(updated_at, id)` keyset	Trigger-maintained columns can be missing on bulk updates
API	`updatedSince` filter + webhooks	Webhooks never advance the watermark

The database caveat is worth a sentence. An updated_at column maintained by an application-level trigger is not updated by a bulk UPDATE ... WHERE issued by a DBA, or by a data migration. The reconciliation pass is what covers it.

-- Reconciliation: compare the full id set, periodically. Catches everything
-- the incremental path missed, whatever the reason.
with source_ids as (select id::text as ref from tickets where closed_at is not null),
     index_ids  as (select source_ref as ref from documents
                     where source = 'ticket_notes' and deleted_at is null)
select 'missing_from_index' as issue, ref from source_ids
 except all select 'missing_from_index', ref from index_ids
union all
select 'orphaned_in_index', ref from index_ids
 except all select 'orphaned_in_index', ref from source_ids;

Run it weekly. It is cheap relative to a full ingest and it is the only thing that catches a systematically broken change signal — the failure mode where incremental sync has been running happily and seeing nothing for a month.

Cost, measured

Atlas's nightly run, after incremental sync:

	Full ingest	Incremental
Documents discovered	2,400	2,400
Fetched	2,400	41
Parsed	2,400	41
Re-embedded	2,400	12
Metadata-only updates	—	4
Wall time	58 min	90 s
Embedding cost	~$14	~$0.07

Discovery stays at 2,400 because gate 1 needs to see every document to know which changed. That is one listing pass and it is the floor on an incremental run's cost — which is why chapter 26's updatedSince filter matters so much for APIs, where discovery is the expensive part rather than the free part.

The gap between 41 fetched and 12 re-embedded is gate 2 earning its place: 29 documents were touched without their content changing, and a hash-free pipeline would have paid full price for all of them.

When to force a full re-ingest

Four triggers. Nothing else justifies it.

export const FULL_REINGEST_TRIGGERS = {
  embeddingModelChanged: 'Vectors from different models are not comparable.',
  chunkingParametersChanged: 'Chapter 31. Every chunk boundary moves.',
  parserFixed: 'Documents parsed by the old version hold the old defect.',
  classificationRulesChanged: 'Chapter 14. Metadata-only, but across everything.',
} as const

The third is the one that gets forgotten. Fixing the table detection in the PDF parser improves nothing until the documents parsed before the fix are re-parsed — and the content hash says they are unchanged, because the source has not changed.

// So the hash must include the parser version, or a parser fix is invisible.
export function hashBlocks(blocks: Block[], parserVersion: string): string {
  const canonical = blocks
    .map((b) => `${b.type}:${blockText(b).replace(/\s+/g, ' ').trim()}`)
    .join('\n')
  return createHash('sha256').update(`${parserVersion}\n${canonical}`).digest('hex')
}

Including the parser version in the hash makes a parser fix self-propagating: bump the version, and the next incremental run re-embeds exactly the documents whose parse actually changed. That is a much better property than remembering to trigger a full re-ingest, and it is three characters of code.

The failure mode: the sync that is complete and empty

Atlas's contract sync runs hourly. Someone changes a folder permission so that the integration account loses read access to one subfolder — 400 of the 3,100 contracts.

The document store's list endpoint returns what the caller can see, which is now 2,700 documents. Every one of them is unchanged. The run reports: 2,700 discovered, 0 fetched, 0 re-embedded, 0 failed, ok: true.

The 400 contracts are still in the index, still retrievable, and now frozen. They will never update again, and nothing will ever say so.

The count check from chapter 23 catches the drop — 2,700 against 3,100 is a 13% fall, which is inside the 20% threshold. So it does not.

// Tighten the check for sources whose count should be stable, and make
// "stable" a property of the source rather than one global threshold.
export function assertDiscoveryPlausible(source: SourceDefinition, seen: number, previous: number) {
  // Contracts grow slowly and never shrink by much. Tickets grow daily.
  const tolerance = source.expectedChurn ?? 0.2
  const drop = (previous - seen) / previous

  if (drop > tolerance) {
    throw new IngestAnomaly(
      `${source.id}: discovered ${seen}, down from ${previous} (${(drop * 100).toFixed(0)}%). ` +
      `Expected churn is ${(tolerance * 100).toFixed(0)}%. Check integration permissions.`,
    )
  }
}

// And the stronger version, where the source can tell you its own total.
const reported = await adapter.count?.()
if (reported && Math.abs(reported - seen) > 1) {
  throw new IngestAnomaly(`${source.id}: discovered ${seen}, source reports ${reported}`)
}

⚠️ Note what the second check does not catch: the source's own count is also filtered by what the integration account can see. It catches pagination bugs, not permission changes. Only the historical comparison catches a permission change, which is why both exist and why the threshold has to be per source.

The message naming the likely cause — check integration permissions — is the same pattern as chapter 26's backfill projection. An alert that names the usual culprit gets fixed; one that reports a number gets acknowledged.

Key points

Two gates: timestamps and etags decide whether to fetch; content hashes decide whether to embed.
Add a metadata_only branch. Identical content with a changed classification is a permission bug a hash cannot see.
Capture the window end before the work, and advance the watermark only on a clean run.
Reconcile the full id set weekly. It is the only thing that catches a change signal that has silently stopped working.
Discovery is the floor on an incremental run. For APIs, the updatedSince filter is what makes that floor cheap.
Include the parser version in the content hash, so a parser fix re-embeds exactly the affected documents by itself.
Four triggers justify a full re-ingest: embedding model, chunking parameters, parser fix, classification rules.
Set the discovery-drop tolerance per source. A source that lost 13% of its documents to a permission change passes a global 20% threshold.
Put the likely cause in the alert message.

Next: deletions, and the chunk that outlives its document — the half of sync that nobody tests.