chatbot_v3/docs/adr/0017-synchronous-ingestion-in-the-request-path.md

0017. Synchronous ingestion in the request path

Status

Proposed

Context

ADR-0014 chose RabbitMQ plus a transactional outbox for durable job dispatch, with three process entrypoints (web, outbox publisher, ingestion worker). ADR-0008 made POST /v1/files job-shaped — 202 Accepted with an ingestion_job_id and a non-terminal status — explicitly so that a durable queue could be added later without breaking clients.

That contract was chosen on the assumption that ingestion is slow enough to exceed HTTP timeouts. For the first slice, that assumption does not hold, and paying for it costs a broker, an outbox table, a publisher process, a worker process, and a client that must poll for completion.

What ingestion actually does per file, and how expensive each stage is:

Stage	Kind of work	Cost
Parse (python-docx, csv)	Blocking CPU, sync SDK	Modest; seconds at worst for a large document
Chunk (fixed-size, ADR-0004)	Blocking CPU, pure Python	Negligible
dense_nomic, dense_openai	Async network I/O	Dominant by wall-clock, but batchable and concurrent
sparse (own BM25 pipeline, ADR-0005)	Blocking CPU	Modest, GIL-bound
Qdrant upsert	Async network I/O	Modest, batchable

The dense embedding stage is the largest term, but it is I/O, not computation this process performs: both embedders take arrays of inputs, and independent batches can be in flight at once. A few hundred chunks is a small number of batched requests, issued concurrently — not a serial per-chunk round trip.

late_interaction (jina-colbert-v2) is the one genuinely heavy ingest-time stage: a document-side multivector per chunk from a local GPU model requiring flash_attn. ADR-0001 defines the vector now to avoid collection recreation but does not require it to be populated, and plan 001 puts the reranker and GPU deployment explicitly out of scope for the first slice. It is therefore not computed during synchronous ingestion, and enabling it is a trigger to revisit this decision rather than a cost this decision has to carry.

Decision

POST /v1/files performs ingestion inline and returns a terminal result. No broker, no outbox, no queue, no worker process, and no polling by the client.

POST /v1/files
  -> authenticate, resolve tenant, validate the upload
  -> 201 Created { file_id, ingestion_job_id, status: "succeeded", chunks_indexed }

ADR-0008's job-shaped 202 Accepted contract is superseded by this ADR; ADR-0014 remains Superseded by 0017 and is the design to adopt if the triggers below are hit. ingestion_jobs is kept exactly as ADR-0009 defines it — it is now a record of an ingestion attempt rather than a queue entry, and it is what makes failures inspectable and re-ingestion idempotent.

Request shape

The request runs in three phases, and the phase boundaries matter:

1. txn A (short):  insert source_files
                   insert ingestion_jobs(status='running', started_at=now())
                   commit  -- release the connection
2. no transaction: store bytes in MinIO
                   parse + chunk            (threads)
                   embed dense + sparse     (async, batched, bounded)
                   upsert Qdrant points     (deterministic ids)
3. txn B (short):  update ingestion_jobs -> succeeded/failed + counters
                   append ingestion_job_events
                   commit

Never hold a Postgres session or transaction open across phase 2. A slow upload would otherwise pin a pool connection — and an idle-in-transaction row lock — for the entire ingestion. Take a session, commit, release, and take a fresh one for phase 3. This is ADR-0012's "one session per unit of work" applied to a request that contains two units.

Committing the job row before the work means a request that dies mid-ingestion still leaves durable evidence: a running job that never reached a terminal state, discoverable by age.

Embedding is concurrent and batched, not serial

Chunks are embedded through the async embedding ports, batched per provider limits, with independent batches in flight concurrently and bounded by a semaphore:

limiter = asyncio.Semaphore(settings.ingestion.embed_concurrency)

async def embed_batch(batch: Sequence[str]) -> list[Vector]:
    async with limiter:
        return await embedder.embed(batch)

batches = chunk_into_batches(texts, size=settings.ingestion.embed_batch_size)
vectors = flatten(await asyncio.gather(*(embed_batch(b) for b in batches)))

Rules:

• Batch before you parallelize. Both dense embedders accept arrays; sending one request per chunk wastes far more time than concurrency can recover.
• Bound concurrency with a semaphore, never an unbounded gather over every batch. The limit exists for the provider's rate limits and for the self-hosted nomic server's capacity — past saturation, extra concurrency just moves the queue somewhere you cannot see it.
• Retry 429/transient failures with backoff inside the request's overall timeout budget, not beyond it.
• The two dense embedders are themselves independent and run concurrently with each other.

Blocking work runs on threads

Parsing (python-docx, csv), chunking, hashing, the sparse BM25 pipeline, and the synchronous minio SDK are blocking. They run via anyio.to_thread.run_sync, with an explicit anyio.CapacityLimiter so ingestion threads cannot exhaust the pool Starlette uses for sync route handlers and dependencies:

chunks = await anyio.to_thread.run_sync(parse_and_chunk, raw_bytes, limiter=ingestion_limiter)

Calling any of them directly from an async def service is a defect: one large python-docx parse would block every concurrent request in the process.

The request is bounded, and says so when it cannot finish

Synchronous ingestion means the client's timeout is now the system's deadline. Three bounds, all configured, all enforced server-side:

• INGESTION_MAX_UPLOAD_SIZE_MB (and a max chunk count) reject work that is obviously too large before any of it starts — a 413, not a timeout.
• INGESTION_TIMEOUT_SECONDS bounds the whole of phase 2. On expiry the job is marked failed with a timeout error code in phase 3, and the response is 504. A timeout must never leave the job stuck in running.
• INGESTION_MAX_CONCURRENCY bounds how many ingestions run in the process at once. Over the limit, the request is rejected with 503 and Retry-After rather than queued behind an unbounded wait — a queue that the client is blocked on is the worst of both designs.

Document the deployment consequence: proxy, load balancer, and client read timeouts must all exceed INGESTION_TIMEOUT_SECONDS, or the client will give up on work that is still succeeding.

Re-running an ingestion stays safe

The idempotency requirements survive, because a client that times out will retry, and phase 2 has no transaction protecting it:

• Point ids are deterministic from file_id + chunk_index (ADR-0001), so a retried upload upserts rather than duplicates.

• Identical uploads are recognized by (tenant_id, domain, content_sha256) and return the existing file/job rather than re-ingesting (plan 001).

• tenant_id comes from AuthContext, never from the request body.

• A terminal job is never transitioned back to running.

• A failed attempt never removes content from a working index. The soft-delete sweep that retires a shortened file's leftover points runs only after every upsert in the attempt has succeeded.

  This is deliberately weaker than "replacement happens only after a successful attempt", which an earlier revision of this ADR claimed. That guarantee is not achievable alongside ADR-0001's deterministic point ids: those ids are exactly what makes a retry idempotent, and they also mean a re-ingestion overwrites points in place, so a crash partway through leaves a prefix updated and the remainder still on the old content. Buying literal atomicity would mean generation-suffixed ids and an activation flip, which contradicts ADR-0001 and ADR-0002's stable point ids. Staging the new points as is_active=false and flipping them on success is strictly worse — the in-place overwrite would deactivate the previously live points, silently emptying a working index if the attempt were interrupted.

  What holds instead: the index is never emptied, never partially deleted, and a retry converges — deterministic ids rewrite every point and the sweep re-runs, reaching the exact correct state.

Failures are HTTP failures

There is no dead-letter queue and no retry loop. A failed ingestion marks the job failed with an error code and a terminating ingestion_job_events row, and returns the corresponding status code (400 for an unparseable file, 413 too large, 503 at capacity, 504 on timeout, 502 for an embedder failure). The client decides whether to retry, which is the correct owner of that decision when the client is synchronous.

GET /v1/files/{file_id} stays

It reports the stored job status. It is no longer a polling mechanism for the upload, but it remains how an operator inspects a past ingestion, and how a client that lost its connection mid-upload discovers what happened.

When to revisit

Return to a queue — ADR-0014's design, or an in-process Postgres-claimed runner as an intermediate step — when any of these becomes true:

• late_interaction document vectors are populated at ingest (GPU, per-chunk model inference: this alone is likely sufficient);
• typical ingestion approaches INGESTION_TIMEOUT_SECONDS, or 504/503 rates stop being negligible;
• file types arrive that are large or slow enough to be unbounded (bulk XLSX, scanned PDFs with OCR);
• ingestion load starts degrading chat/retrieval latency in the same process;
• ingestion needs to be retried automatically rather than by the caller.

Track p95 ingestion duration, 503/504 counts, and the count of running jobs older than the timeout. Those are the trigger metrics.

Consequences

Positive

• The client gets its answer in one call. No polling, no job-status endpoint in the happy path, no "queued" state to explain to a frontend.
• The first slice needs Postgres, MinIO, and Qdrant only — no broker, no outbox table, no publisher or worker process, no exchange/queue/DLX provisioning.
• Failures reach the caller directly, with a real status code and message, instead of being discovered later by polling a job row.
• One process to run, deploy, and reason about; docker compose up is the whole application.
• Batched, bounded-concurrent embedding is a property worth having regardless of where ingestion runs — it carries over unchanged into a queued design.

Negative

• The request's duration is now a product constraint. Proxy/client timeouts become deployment configuration that can silently break uploads.
• Ingestion competes with chat/retrieval for CPU, threads, and connections in the same process, and a burst of uploads degrades API latency for everyone.
• No automatic retry: a transient embedder failure surfaces as a 502 and depends on the caller to retry.
• A client disconnect does not cancel the work, and leaves a job that can sit in running until observed by age.
• Capacity rejection (503) is a worse experience than queueing for callers who would rather wait — accepted deliberately, because a blocked client waiting on a hidden queue is worse still.
• Reversing this decision changes the HTTP contract (201 with a terminal status becomes 202 with a pending one), so clients would have to change. This is the real cost of the decision, and the reason the trigger list above is explicit.

Alternatives Considered

• Job-shaped 202 Accepted with a queue (ADR-0008 as written, ADR-0014): rejected for now. It is the right design once ingestion is genuinely slow or needs automatic retry, and the trigger list says when to adopt it. Building it first would add a broker, an outbox, and two processes to make a seconds-long operation asynchronous.
• In-process Postgres-claimed job runner with 202: rejected as the current step, but it is the natural intermediate — it keeps the single process and the single deployment while decoupling the request from the work. It is the first thing to reach for when the triggers fire, before adopting a broker.
• Serial per-chunk embedding: rejected. It is the version of synchronous ingestion that genuinely would exceed HTTP timeouts, and it is avoidable with batching plus bounded concurrency.
• Unbounded asyncio.gather over all batches: rejected. It converts a large upload into a rate-limit burst against the provider and an unbounded memory spike locally; the semaphore is what makes the concurrency safe.
• Run the blocking stages on the event loop directly: rejected. A single large python-docx parse would stall every concurrent request in the process.
• BackgroundTasks after returning 201: rejected as the worst of both — the client is told the work succeeded before it has, with no durable record and no retry if the process restarts.
• Populate late_interaction during synchronous ingestion: rejected for this slice, consistent with plan 001's scope. Per-chunk GPU model inference is the stage that makes ingestion unbounded; adding it is a trigger to revisit this ADR, not something to absorb into a request.