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