Skip to content

Ingest state machine

The document pipeline drives every file through an explicit state machine. A crash mid-file resumes from the last committed step instead of re-running everything. This document is the engineer reference: step diagram, error categories, retry semantics, and how to add a new step.

Related code:

Data model

Every ingestion creates one DocumentIngestionRun. Each file in the run becomes one DocumentIngestionRunItem. Each step transition emits one DocumentIngestionRunEvent with the duration and (on failure) the error category; per-call cost lives in the ModelUsageEvent ledger, attributed to the run, item, and step.

DocumentIngestionRun (kind, status, item_count_*)
  └── DocumentIngestionRunItem (step, retry_count, last_error_kind)
        └── DocumentIngestionRunEvent  (step_started | step_completed | step_failed | retry_scheduled)

Steps

step                  SourceFile.status   ConversationAttachment.status
ENQUEUED              uploaded            extracting
FETCHED               extracting          extracting
CONTENT_EXTRACTED     extracting          extracting
CHUNKED               extracting          extracting
CHUNKS_PERSISTED      extracting          available
  (EMBEDDED || ENTITY_EXTRACTED)
EMBEDDED              extracting          available
ENTITY_EXTRACTED      extracting          available
ENTITIES_PERSISTED    extracting          available
INDEXED               extracted           ready
DONE                  extracted           ready

  Any state → FAILED (or ABANDONED, via the run reconciler)
FAILED                failed              failed
ABANDONED             failed              failed

The parallel group runs concurrently via DocumentIngestionStateMachine.run_parallel_steps and rendezvouses before ENTITIES_PERSISTED. The worker drives per-step transitions: run_pipeline (pipeline/runner.py) wraps each step function and advances the item through DocumentIngestionStateMachine.run_step, so every step gets its own started/completed/failed events and idempotent resume. The order itself is declared exactly once, as STAGES in services/document_ingestion/steps.py: the state machine flattens it into STEP_ORDER and the runner zips it with the step functions to build PIPELINE, refusing to import if its function table disagrees.

ENTITY_EXTRACTED additionally writes its own EntityExtractionSignal row recording what that attempt produced. It is keyed on (source_file, run_item) so a resumed step updates the row rather than adding a second one, and a write failure is logged without failing the ingest - telemetry about extraction must not be able to break extraction. The row is written two steps before entities are persisted, so a later failure can leave a row describing documents that were never written. See document-pipeline.md §7.

What the file and the attachment show

The run item's step is the only writer-owned record of where a file is. SourceFile.status and ConversationAttachment.status are the coarser copies the folder list and the chat composer read, and they follow the item: every place that moves item.step writes both columns through file_status.write_statuses_for_step in the same transaction, using the table in the step diagram above. Nothing under pipeline/ and nothing in the worker writes either column any more.

A file can have several items - a reprocess, a scrape re-enqueue and the connector retry each make one - so the writer follows the same rule the readers do: only the file's newest item writes its columns, and an older item still moving writes nothing. That is what stops the daily reconciler, abandoning an orphan from a lost message, from overwriting the file its successor finished.

Three things the mapping does not own. The pre-enqueue half of SourceFile.status - uploading while the browser is still sending bytes, uploaded once they have landed - is written by the upload paths before any item exists. ConversationAttachment.status = removing is the cooperative-cancellation marker the remove paths set; the mapping skips those links so a remove in flight is never overwritten by a step that is still finishing. And the reprocess paths restore a file's prior status when their own enqueue fails (process_content, extraction_review), which is an undo of a pre-enqueue write, not a step.

Two writes in connector_file_ingest.py are deliberate exceptions of a different kind. _reenqueue_failed_copy flips failed to uploaded as a compare-and-set, so that exactly one of several conversations retrying the same failed copy publishes the retry; the enqueue that follows writes the same value through the mapping. If that enqueue fails before it has made an item, the same function puts failed back, because a copy left at uploaded with a failed newest item could never win the flip again. _await_answerable re-asserts the ground truth it has just read onto the one link it is waiting on, which costs nothing and closes the gap if that link was created stale.

An abandoned item reads failed on both columns, with the reconciler's reason. The item stays abandoned rather than failed, so the Failed Items Inbox and the failed-items health signal still exclude it - but from the user's side nothing about that file will ever finish, and uploaded forever would say the opposite.

A retryable failure moves neither column. The item records the error and bumps retry_count; the file stays wherever its last successful step put it, because the attempt is not over. Only a terminal failure writes failed, and it writes the step's message as the reason on the file and on every live link.

A link to a shared connector-file copy is the one place that has to answer "what should this link read?" without having just moved the item. link_status.link_status_for_file answers it from the copy's newest run item - highest pk, because a reprocess creates a new item - mapped through the same table. A copy with no item at all reads extracting, which covers the window between the copy row being created and its ingest being published; the exception is a copy the enqueue-failure rollback left failed, which reads failed with the file's own reason so the next load retries it.

Because the file's status is written when the state machine records INDEXED, and finalize_indexes deletes superseded files inside that step, there is a moment between the delete and the transition where neither the victims nor the new file reads extracted. Nothing depends on the new file's status during that window: what keeps the delete safe is the victim filter excluding the new file by id.

Error categories

Every external call returns one of five categories. The state machine routes based on the category:

error_kind Worker reaction
"" (status=ok) Advance to the next step.
RateLimited Wait on the rate budget, retry the same step.
Transient Exponential backoff, retry up to max_retries.
SchemaInvalid Terminate to the Failed Items Inbox. Do not retry - same input → same output → token waste.
Fatal Terminate to the Failed Items Inbox. Auth failures, missing input, programmer bugs.

The reaction matrix is enforced inside DocumentIngestionStateMachine.run_step. Categories are produced by result_from_exception (which maps exception subclasses) or directly by Lambda handlers (which embed the category in their return envelope).

Idempotency

run_step(target, callable) is idempotent in its durable writes. If item.step is already at-or-past target, the callable still runs - rebuilding the in-memory StepContext later steps read from - but no events are emitted and the step is not moved, so durable rows are never duplicated. This is the resume-from-crash mechanism: when SQS redelivers a message, the worker re-runs the cheap earlier steps and picks up the durable record where it left off.

The transition itself is atomic via SELECT FOR UPDATE + update() inside a transaction. The SQS visibility timeout is the outer lock.

Adding a new step

  1. Add the value to DocumentIngestionStep in models/document_ingestion.py.
  2. Insert it into STAGES in services/document_ingestion/steps.py - as its own stage, or inside an existing parallel group.
  3. Generate a migration (makemigrations picks up the choices change).
  4. Map the step to its async function in _STEP_FNS (pipeline/runner.py); the runner fails at import until the table matches STAGES.
  5. Add a unit test covering the happy path + all four failure categories for the new callable.

Cross-references