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, started_at, completed_at, item_count_*)
  └── DocumentIngestionRunItem (step, outcome, retry_count, last_error_kind)
        └── DocumentIngestionRunEvent  (event_type: DocumentIngestionRunEventType, step: DocumentIngestionStep, duration_ms)
              └── DocumentIngestionStepSpan  (name: DocumentIngestionSpanName, total_ms, count)

A run is created running and stays there until something closes it, so there is no opening state to move out of and no mark_run_started. The clock rule is in the schema, not in the writers: a terminal run has a completed_at, a running one does not, and started_at is never null.

duration_ms follows the event type, and a check constraint says so: step_completed always carries a duration; step_failed carries one when the failure ends a measured attempt, and terminate_item's administrative failure rows carry none; step_started and retry_scheduled never carry one. That is what lets the timing queries read a duration off an event without guessing whether the column means "not measured" or "measured as nothing".

step is where the file reached and outcome is how its ingest ended; the two are separate columns and six check constraints hold step, outcome, last_error_kind, and last_error_message to each other and to their enums. A failure or an abandonment writes the outcome and leaves the step alone, so an item that stopped mid-pipeline still says where it stopped.

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

The outcome answers first, whatever step the item had reached:

outcome               SourceFile.status   ConversationAttachment.status
IN_PROGRESS           (the step decides)  (the step decides)
DONE                  (the step decides)  (the step decides)
FAILED                failed              failed
ABANDONED             failed              failed

IN_PROGRESS is every item that can still move, including one between a retryable failure and its next attempt. DONE is set in the same write that commits DONE as the step, and the row constraint refuses any row where those two disagree. FAILED is a terminal failure, and ABANDONED is the run reconciler settling an item whose message is provably gone.

The parallel group is gathered by the runner: run_pipeline starts a run_step call for each member that a resume did not skip, awaits them all, 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. Text too long for the extraction model does not fail the step: the row records the too_long outcome, no entities are stored, and the item goes on to DONE. See document-pipeline.md §7.

What the file and the attachment show

The run item's (step, outcome) pair is the only writer-owned record of where a file is and how its ingest ended. 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 either column writes both status columns through file_status.write_statuses_for_item in the same transaction, using the two tables 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.

A failure found where no item speaks for the file goes through the module's second door, file_status.write_failure_for_file. Those failures are a publish that never made an item, an operator's verdict from the admin set-status action, and the restore of a failed prior. It writes the file and every link except removing ones in one transaction, exactly as the item path does, so the write itself never leaves a link reading healthy beside the file it failed. Its two guards make the write conditional: only_while is a compare-and-set on the file's current status, and unless_item_after skips the write when an item newer than the given pk exists, because that item speaks for the file. Either guard refusing writes nothing, links included. A later read can still disagree with that write on the connector path. link_status_for_file derives a connector copy's link status from the copy's newest run item, so after an operator marks the copy failed through the admin action, the next _await_answerable pass or a new conversation's link reads what that item maps to, such as ready, while the older links read failed. The links to one file agree again once a newer item settles them. test_failure_writer_pin parses every production module under apps/ and fails on a failure written into either table anywhere but file_status.py.

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. A failed prior goes back through write_failure_for_file, so the file's links read the restored reason too; any other prior goes back through a compare-and-set update. In extraction_review and process_content alike, both restore branches only write while the file is still uploaded, so a publish failure's failed stands, and so does a worker's extracting.

A failed publish is a mapping write of its own: trigger.enqueue terminates the item it has just created, so the file and every live link read failed with "The file could not be queued for processing." What it raises then tells the caller which failure this was - IngestPublishError for a queue that heals by itself, meaning a network fault, a 5xx or a throttle, and the original exception for everything else, such as missing credentials or a queue that does not exist. reprocess_from_raw catches either one per page, counts it and carries on, so one refused publish leaves the rest of the host queued, and the command exits non-zero when any page failed. It stops the run after three refusals in a row, because it publishes on the retrying client, where a queue that is down costs about four minutes per page, and it reports how many pages it never attempted.

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 through write_failure_for_file with unless_item_after set to the newest item before the retry, because a copy left at uploaded with a failed newest item could never win the flip again. The question is whether the retry made an item, not what the file reads: an item can exist without having been terminated, when terminate_item itself raises, and that file is left to the item. _await_answerable re-asserts a healthy ground truth onto the one link it is waiting on, clearing any reason on it, which costs nothing and closes the gap if that link was created stale. A failed ground truth goes through write_failure_for_file with only_while=failed, so a copy another load has just claimed for a retry is not flipped back and its links are left for the retry's item.

An abandoned item reads failed on both columns, with the reconciler's reason. Its outcome stays abandoned rather than failed, so the failed-items health signal still excludes it - but from the user's side nothing about that file will ever finish, and uploaded forever would say the opposite. Abandonment can settle an item at any step, so an item abandoned past chunks_persisted keeps its chunk rows, which stay searchable because retrieval filters on access scope and never on SourceFile.status. The Failed Items Inbox lists abandoned items alongside failed ones, so an operator requeues one from the step it reached.

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.

The worker publishes the retry before it deletes the delivery it holds, so a publish that fails leaves the original to come back after the visibility timeout with retry_count already counted. A delete that fails after the publish leaves two messages for one item, which the second-consumer rules below absorb: a live item spends one extra retry, and a done item logs the discard. Artifact export and render jobs follow the same order for their own retries.

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

Both columns that store a kind - DocumentIngestionRunItem.last_error_kind and DocumentIngestionRunEvent.error_kind - carry a membership CheckConstraint on DocumentIngestionErrorKind.values, so a value outside the enum fails the write instead of reaching a reader. The event row's other two columns are held the same way: event_type against DocumentIngestionRunEventType and step against DocumentIngestionStep, so a reader can rank an event's step along STEP_ORDER without filtering out names from older builds first.

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 Republished with a wait, retried up to max_retries; the retry's calls wait on the rate budget.
Transient Republished with a wait, retried 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. A corrupt file, a missing dependency, a request the provider rejected.

The reaction matrix is enforced inside DocumentIngestionStateMachine.run_step. max_retries is one budget for the whole item, counted across every step, and the worker republishes a retry with a wait that depends on the queue: 15 s then 60 s on the interactive queue, and 60 s then 300 s on the bulk queue, whatever the failure's kind. Categories are produced by result_from_exception, which maps exception subclasses. An exception a step raises uncategorized is mapped by the runner's _categorise first. An instance of UNRETRYABLE_EXCEPTION_TYPES (ImportError, and litellm's ContextWindowExceededError and ContentPolicyViolationError) is Fatal on its first attempt, because no retry can change it, and anything else is logged, counted and filed by the shared as_categorized_error rule plus the runner's own schema-text rule, which default to Transient (document-pipeline.md §6). See document-pipeline.md §6 for the rule and the retries litellm makes inside each attempt.

Idempotency

run_step(target, callable) is idempotent in its durable writes. If item.step is already at-or-past target, whatever the runner hands it 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.

Which callables the runner hands over on a resume is decided before the loop, from the rows. pipeline/rebuild.py fills the context from the chunk rows and SourceFile.parsed_file for an item at or past chunks_persisted, and returns the steps those rows account for; the runner skips exactly those and runs everything else, in hydration mode where the item is already past it. So a redelivery after a crash picks up the durable record where it left off without re-downloading the source or re-running extraction, and the state machine's own contract is unchanged: it runs what it is handed.

run_step also times the callable and records what happened inside it. It installs a span tracker for the duration of the call (services/llm_infra/span_tracker.py), and the pipeline functions open a named span(...) around each piece of work - the OCR call, the parser, each LLM call. The tracker's per-name totals are written as DocumentIngestionStepSpan rows in the same transaction as the step_completed or step_failed event, so a step's duration can be read apart into where the time went. A failed attempt records its spans too: the time an OCR call burned before the step gave up is still time the document cost.

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

Redelivery stops at a settled item. run_step refuses to advance one (TerminalStepError), so before entering the pipeline the worker checks item.is_settled - that is, outcome != in_progress - and deletes the message instead of letting the raise escape to the DLQ. The reconciler settles an item whose message it cannot see, and an in-flight message is one it cannot see, so this pairing is reachable in normal operation rather than only after a crash.

The entry guard is enough because the reconciler abandons only an item whose message it has proven gone, and a message being worked on is in flight, which the queue depth counts. The depth sums three SQS counts, visible, in flight and delayed, so a retry waiting out its delay counts too. A counter that under-reports is the one way a live item can be abandoned underneath its own run, and that ends in a refused advance rather than duplicated work: _advance_to re-reads the outcome under the row lock and writes nothing, run_step raises TerminalStepError, and the entry guard discards the redelivery. A branch that finishes after a sibling in its parallel group failed terminally is covered by the same re-read.

An item can also be moved by a second consumer of its own message, and the writers refuse to rewind it. The document path never calls SQS change_message_visibility; only the artifact jobs run a heartbeat that does. So a document delivery that outlives the queue's visibility timeout is handed to a second worker while the first is still inside run_pipeline, and each worker holds its own in-memory copy of the item. Both _advance_to and _record_failure therefore read the row's step under the same lock as their write and refuse when the step they were handed belongs to an earlier stage than the one the row has reached. "Earlier" is a stage, not a STEP_ORDER position, because the parallel group's two members legitimately commit in either order and the second one to finish writes the lower position. A refusal logs at WARNING with both steps and the item id, increments ingest.advance_refused_behind_row, and marks the state machine overtaken; run_pipeline then returns None, which the worker treats as a clean exit that deletes the message and records a success. That is deliberate: the item is healthy and the other consumer is still carrying it, so the log line and the metric are the only trace. The worker's own terminal write through terminate_item is exempt, because a resume-hydration failure names the step whose rebuild failed, which is behind the row by definition.

The reconciler settles an item under the same discipline. _mark_item_abandoned locks the row, updates only while the item is still in_progress, and writes the file statuses from the step the locked row holds, so a worker that advances the item between the candidate select and the write keeps it.

A done item is the one case worth a second look, and the worker logs it at WARNING. Advancing to done, bumping the run's item_count_completed and closing the run are three separate transactions, and the message is deleted after all three; a worker that dies in between leaves a done item whose counter was never bumped. mark_run_finished is a pure counter read, so that run never closes on its own and shows up only as a stuck run on the Pipeline Manager. The same WARNING also fires for the harmless duplicate a retry leaves when the original's delete fails after the retry was published, so the log line names both causes.

Resuming a failed item

The Failed Items Inbox re-enqueue publishes a new manual_reprocess run whose item starts at the failed item's step, through trigger.enqueue(start_step=...). The failed item is left exactly as it is, as the record of what went wrong, and the new run's parameters carry resumed_from_step. An item that is no longer its file's newest is skipped and named in the action's summary, because its new item would immediately stamp a mid-ladder status over a successor's finished work.

Starting past enqueued changes both the record and the work. The step after start_step is the first to get its own event pair and cost rows, and a start at or past chunks_persisted rebuilds the context from the rows the earlier steps already wrote instead of re-running them. A resume from chunks_persisted therefore re-downloads nothing, runs no extractor and re-writes no chunk rows: fetched, content_extracted, chunked and chunks_persisted are skipped outright. embedded is skipped only when the item is at or past it and every chunk row of the file holds a vector, because after the parallel group step is whichever member finished last. entity_extracted and entities_persisted are skipped only together and only at or past entities_persisted: EntityBatch is durable nowhere earlier, so an item resumed at embedded or entity_extracted still spends its entity-extraction call to rebuild the batch. Any step the rows cannot account for runs in hydration mode as before, emitting nothing and moving nothing.

A resume whose rows are gone - a parsed_file that predates the writer, chunk rows an operator deleted - falls back to running every callable, which is the old behaviour: correct, and expensive. That path logs a WARNING naming the item and what was missing and increments ingest.resume_rebuild_fallback, so it is visible rather than silent, and the file still finishes instead of being sent to the inbox for a row that was never a contract.

The user sees the resume immediately: the new item writes the file and every chat link to whatever its start step maps to, so a resume from chunks_persisted reads extracting on the file and available on the links rather than dropping back to uploaded.

Adding a new step

  1. Add the value to DocumentIngestionStep in models/document_ingestion.py. It is a position, never a verdict - how an ingest ended is DocumentIngestionOutcome.
  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, which the documentingestionrunitem_step_is_a_known_member constraint enforces).
  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