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.

```text
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:

```text
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:

```python
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:

```python
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`.
- Qdrant points from a failed attempt do not replace the previous successful
  index; replacement happens only after a successful attempt.

### 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.