docs(architecture): adopt inline synchronous ingestion (ADR-0017)
Why: - ingestion is inline in the request instead of dispatched through RabbitMQ/outbox/worker; ADR-0014 is superseded (not deleted) and named as the design to adopt once ingestion needs to move off the request path. Changes: - new ADR-0017 plus amendments to every ADR/plan that referenced the job-shaped/broker contract, so none silently contradict it. Impact: - no broker, outbox, or worker code; rabbitmq test marker removed.
This commit is contained in:
@@ -4,9 +4,8 @@
|
||||
|
||||
This plan turns the accepted architectural direction in the ADRs into the first
|
||||
working product slice: a tenant-scoped CSV upload is stored in MinIO, represented
|
||||
by durable Postgres records, dispatched through RabbitMQ using a
|
||||
transactional outbox, processed by a separate worker, and indexed as Qdrant
|
||||
points.
|
||||
by durable Postgres records, parsed/chunked/embedded inline in the request
|
||||
(ADR-0017), and indexed as Qdrant points before the response returns.
|
||||
|
||||
This is an implementation plan, not an Architecture Decision Record. ADRs explain
|
||||
why major technologies and boundaries were chosen; this document defines the
|
||||
@@ -19,10 +18,9 @@ The first vertical slice uses these responsibilities:
|
||||
| System | Responsibility |
|
||||
|---|---|
|
||||
| FastAPI | HTTP boundary, validation, authentication, tenant derivation, and job creation. |
|
||||
| Postgres | Tenant/auth data, source-file metadata, ingestion job state/progress, audit, and transactional outbox events. |
|
||||
| Postgres | Tenant/auth data, source-file metadata, ingestion attempt records/progress, and audit. |
|
||||
| MinIO | Private source-file bytes and retained derived ingestion blobs. |
|
||||
| RabbitMQ | Durable delivery of ingestion and maintenance work. |
|
||||
| Ingestion worker | Parsing, chunking, embedding, ingestion-generated Chunk/Point CRUD, and job status updates. |
|
||||
| Ingestion service | Parsing, chunking, embedding, ingestion-generated Chunk/Point CRUD, and job status updates — inline in the request. |
|
||||
| Qdrant | Tenant-filtered generated chunks and their vectors/payloads. |
|
||||
|
||||
The controlling ADRs are:
|
||||
@@ -35,9 +33,11 @@ The controlling ADRs are:
|
||||
resource lifetime, dependency injection, and explicit transaction ownership.
|
||||
- [ADR-0013](../adr/0013-s3-compatible-object-storage-with-minio.md): MinIO object
|
||||
storage boundary.
|
||||
- [ADR-0014](../adr/0014-durable-job-dispatch-with-rabbitmq.md): RabbitMQ,
|
||||
transactional outbox, separate workers, and worker-owned ingestion Chunk/Point
|
||||
CRUD.
|
||||
- [ADR-0017](../adr/0017-synchronous-ingestion-in-the-request-path.md):
|
||||
inline ingestion, batched/bounded-concurrent embedding,
|
||||
`anyio.to_thread.run_sync` for blocking work, and request bounds. It supersedes
|
||||
[ADR-0014](../adr/0014-durable-job-dispatch-with-rabbitmq.md), which remains
|
||||
the design to adopt when a broker becomes necessary.
|
||||
|
||||
The cited ADRs are currently proposed. Treat them as the implementation baseline
|
||||
only after the project owner accepts them; code should not silently diverge from
|
||||
@@ -50,13 +50,14 @@ them.
|
||||
- `POST /v1/files` for authenticated tenant-scoped **CSV** upload.
|
||||
- File validation, size limits, content hashing, and streaming upload to MinIO.
|
||||
- Alembic-managed Postgres schema for the minimal tenant/auth, source file,
|
||||
ingestion job, job event, and outbox records needed by this slice.
|
||||
- Transactional outbox publication of `ingestion.job.created` to RabbitMQ.
|
||||
- A separate ingestion worker process with a durable RabbitMQ consumer.
|
||||
ingestion job, and job event records needed by this slice.
|
||||
- Inline ingestion in `POST /v1/files`, with batched/bounded-concurrent
|
||||
embedding, thread-offloaded parsing, and enforced size/timeout/capacity
|
||||
bounds.
|
||||
- CSV parsing and deterministic chunk creation.
|
||||
- Tenant-filtered Qdrant point upserts using deterministic point identifiers.
|
||||
- Job status/progress persistence and `GET /v1/files/{file_id}` status lookup.
|
||||
- Structured correlation logging at HTTP, outbox, and worker ingress.
|
||||
- Structured correlation logging at HTTP and ingestion-stage boundaries.
|
||||
- Automated tests for the critical state transitions, redelivery, and tenant
|
||||
boundaries.
|
||||
|
||||
@@ -70,31 +71,40 @@ them.
|
||||
ingestion.
|
||||
- Full tenant erasure and hard-deletion workflow.
|
||||
- A public download API or presigned object URLs.
|
||||
- Exactly-once end-to-end processing. The worker must instead be safe under
|
||||
at-least-once delivery.
|
||||
- A message broker, transactional outbox, job queue, and separate
|
||||
worker/publisher processes (ADR-0014, deferred by ADR-0017).
|
||||
- `late_interaction` (jina-colbert-v2) document vectors at ingest — populating
|
||||
them is ADR-0017's primary trigger to move ingestion back off the request.
|
||||
- Exactly-once end-to-end processing. Job execution must instead be safe when a
|
||||
job runs more than once.
|
||||
|
||||
## Required invariants
|
||||
|
||||
The implementation must preserve these rules from the ADRs:
|
||||
|
||||
1. `tenant_id` is derived from trusted authentication context; it is never
|
||||
accepted from the upload body, query parameters, MinIO metadata, or a broker
|
||||
message as authority.
|
||||
accepted from the upload body, query parameters, MinIO metadata, or a
|
||||
dispatch payload as authority.
|
||||
2. MinIO stores bytes; Postgres stores metadata, lifecycle state, job progress,
|
||||
audit records, and dispatch intent.
|
||||
3. Broker messages contain stable IDs and correlation metadata only. They never
|
||||
audit records, and the queue.
|
||||
3. Dispatch payloads contain stable IDs and correlation metadata only. They never
|
||||
contain file bytes, extracted text, chunks, embeddings, secrets, raw prompts,
|
||||
or raw model output.
|
||||
4. The worker reloads the job and source-file records from Postgres before doing
|
||||
tenant-scoped work.
|
||||
5. The worker, not the HTTP publisher, performs parsing, chunking, embedding, and
|
||||
generated Qdrant Chunk/Point CRUD.
|
||||
4. Ingestion works from the persisted job and source-file records, not from
|
||||
request-supplied values.
|
||||
5. Parsing, chunking, embedding, and generated Qdrant Chunk/Point CRUD live in
|
||||
the ingestion application service, not in the route handler. Blocking work
|
||||
runs through `anyio.to_thread.run_sync`, never directly on the event loop.
|
||||
6. Qdrant reads and mutations are tenant-filtered. Ingestion-generated point IDs
|
||||
are deterministic so retrying a job does not create duplicate logical chunks.
|
||||
7. The worker acknowledges a RabbitMQ message only after it has durably persisted
|
||||
the applicable Postgres progress/final state.
|
||||
8. Application clients are built at FastAPI or worker-process startup and closed
|
||||
at shutdown. No mutable clients are opened as import-time globals.
|
||||
7. No Postgres session or transaction is held open across parse/embed/upsert.
|
||||
The job row is committed `running` before the work and updated to a terminal
|
||||
status after it, in a second short transaction.
|
||||
8. Application clients are built at FastAPI startup and closed at shutdown. No
|
||||
mutable clients are opened as import-time globals.
|
||||
9. Embedding is batched per provider limits and concurrency-bounded by a
|
||||
semaphore; blocking work (parse, chunk, BM25, `minio`) runs through
|
||||
`anyio.to_thread.run_sync` with an explicit `CapacityLimiter`.
|
||||
|
||||
## Required decisions before implementing the affected phase
|
||||
|
||||
@@ -120,22 +130,25 @@ rather than becoming an accidental repository behavior.
|
||||
For the first release, file deletion should be soft and job-shaped:
|
||||
|
||||
1. mark the source file as deletion requested/soft deleted in Postgres;
|
||||
2. write a maintenance outbox event;
|
||||
3. have a worker soft-delete the related Qdrant points;
|
||||
2. soft-delete the related Qdrant points inline, recording the attempt;
|
||||
4. retain the MinIO object until an explicit retention or hard-erasure workflow.
|
||||
|
||||
Hard deletion requires a later retention/erasure implementation covering MinIO,
|
||||
Qdrant, and the relevant Postgres data.
|
||||
|
||||
### Broker operations
|
||||
### Ingestion bounds and operations
|
||||
|
||||
Before deploying an environment, define and document:
|
||||
|
||||
- RabbitMQ exchange, queue, binding, and dead-letter-exchange configuration;
|
||||
- queue name, prefetch, manual-ack policy, and DLX retry/backoff;
|
||||
- worker concurrency and resource limits;
|
||||
- outbox polling/publish interval and stuck-event alert threshold;
|
||||
- how failed jobs are inspected, retried, and cancelled.
|
||||
- `INGESTION_MAX_CONCURRENCY` and the thread-pool capacity limiter, and their
|
||||
relation to the process's CPU/memory budget;
|
||||
- `INGESTION_TIMEOUT_SECONDS`, and the proxy/load-balancer/client read timeouts
|
||||
that must exceed it;
|
||||
- `INGESTION_EMBED_BATCH_SIZE` and `INGESTION_EMBED_CONCURRENCY`, sized to the
|
||||
provider's rate limits and the self-hosted embedder's capacity;
|
||||
- alert thresholds for p95 ingestion duration, `503`/`504` rates, and jobs left
|
||||
in `running` past the timeout;
|
||||
- how failed ingestions are inspected and retried.
|
||||
|
||||
These are deployment/runbook settings, not new ADRs unless they change the
|
||||
reliability guarantee or system boundary.
|
||||
@@ -144,15 +157,14 @@ reliability guarantee or system boundary.
|
||||
|
||||
### Phase 1: Foundation and local dependencies
|
||||
|
||||
1. Add typed configuration in `src/config.py` for Postgres, MinIO, RabbitMQ,
|
||||
Qdrant, application limits, and logging.
|
||||
1. Add typed configuration in `src/config.py` for Postgres, MinIO, Qdrant,
|
||||
ingestion bounds, application limits, and logging.
|
||||
2. Populate `.env.example` with non-secret local-development configuration.
|
||||
3. Add application Docker Compose services for Postgres, MinIO, RabbitMQ, and
|
||||
Qdrant. Keep application MinIO buckets/credentials separate from Langfuse
|
||||
infrastructure.
|
||||
3. Add application Docker Compose services for Postgres, MinIO, and Qdrant. Keep
|
||||
application MinIO buckets/credentials separate from Langfuse infrastructure.
|
||||
4. Add direct Python dependencies and lock them with `uv`:
|
||||
SQLAlchemy async/Postgres driver, MinIO/S3 client, aio-pika, Qdrant client,
|
||||
and structured logging dependencies chosen by ADR-0011.
|
||||
SQLAlchemy async/Postgres driver, MinIO/S3 client, Qdrant client, and
|
||||
structured logging dependencies chosen by ADR-0011.
|
||||
5. Add the pytest foundation from ADR-0016: async test configuration, boundary
|
||||
markers, and support for dependency-injected fakes. Add a lifespan smoke test
|
||||
before creating external clients.
|
||||
@@ -166,11 +178,10 @@ run without Docker or live providers.
|
||||
### Phase 2: Database, migrations, and domain contracts
|
||||
|
||||
1. Define SQLAlchemy models and Alembic migrations for the minimum required
|
||||
tables: `tenants`, `api_keys`, `source_files`, `ingestion_jobs`,
|
||||
`ingestion_job_events`, and `outbox_events`.
|
||||
2. Define Pydantic request/response/message schemas, including a versioned
|
||||
`ingestion.job.created` message containing `event_id`, `tenant_id`, `file_id`,
|
||||
`ingestion_job_id`, `request_id`, and `api_key_id`.
|
||||
tables: `tenants`, `api_keys`, `source_files`, `ingestion_jobs`, and
|
||||
`ingestion_job_events`, per ADR-0009.
|
||||
2. Define Pydantic request/response schemas, including the terminal upload
|
||||
response (`file_id`, `ingestion_job_id`, `status`, `chunks_indexed`).
|
||||
3. Implement explicit repositories/services with a request/job-lifetime
|
||||
`AsyncSession`; routes/services own commit/rollback boundaries as specified by
|
||||
ADR-0012.
|
||||
@@ -188,75 +199,75 @@ reads/writes and valid job transitions.
|
||||
2. Implement `POST /v1/files` for CSV only, including streaming-size controls,
|
||||
file-type validation, SHA-256 calculation, and a private MinIO upload using
|
||||
an internal object key.
|
||||
3. In one Postgres transaction, persist `source_files`, create
|
||||
`ingestion_jobs(status='queued')`, and write the corresponding unpublished
|
||||
`outbox_events` row.
|
||||
4. Return `202 Accepted` with `file_id`, `ingestion_job_id`, and `queued` status.
|
||||
3. In one short Postgres transaction, persist `source_files` and create
|
||||
`ingestion_jobs(status='running')`, then commit and release the connection
|
||||
before any parse/embed work.
|
||||
4. Return `201 Created` with `file_id`, `ingestion_job_id`, terminal status, and
|
||||
`chunks_indexed` once ingestion completes.
|
||||
5. Implement `GET /v1/files/{file_id}` with tenant filtering and a public status
|
||||
response that does not expose raw storage credentials or internal artifacts.
|
||||
6. Add cleanup/compensation handling for a MinIO upload that succeeds while the
|
||||
database transaction fails.
|
||||
7. Add unit/API tests for trusted tenant derivation, CSV validation, idempotency,
|
||||
`202 Accepted`, and tenant-scoped status. Add MinIO adapter integration tests
|
||||
the terminal `201 Created` response, and tenant-scoped status. Add MinIO adapter integration tests
|
||||
for server-derived private object paths and compensation behavior.
|
||||
|
||||
**Exit criteria:** an authenticated CSV upload creates a private object, a queued
|
||||
job, and an unpublished outbox event; a tenant cannot retrieve another tenant's
|
||||
file status; the HTTP path does not publish directly to RabbitMQ.
|
||||
**Exit criteria:** an authenticated CSV upload creates a private object and a
|
||||
`running` job row committed before any ingestion work; a tenant cannot retrieve
|
||||
another tenant's file status.
|
||||
|
||||
### Phase 4: RabbitMQ and outbox publisher
|
||||
### Phase 4: Bounded execution primitives
|
||||
|
||||
1. Provision/verify the application-owned RabbitMQ topic exchange, queue, and
|
||||
binding configuration through deployment/bootstrap code rather than route
|
||||
startup side effects.
|
||||
2. Implement an outbox-publisher process that safely claims unpublished events,
|
||||
publishes them with the stable event id using publisher confirms, and records
|
||||
success/failure attempts.
|
||||
3. RabbitMQ has no broker-native message de-duplication; retain idempotency in
|
||||
all consumers as the sole guard against duplicate processing.
|
||||
4. Add monitoring/logging for publish attempts, unpublished-event age, and
|
||||
repeated failures.
|
||||
5. Add unit tests for event claiming and retryable failures, then Testcontainers
|
||||
RabbitMQ integration tests for durable publication, restart/retry, duplicate
|
||||
publication, and message metadata.
|
||||
1. Add the async embedding ports and adapters in `src/infrastructure/embedding/`,
|
||||
with per-provider batching and an `asyncio.Semaphore` bounding in-flight
|
||||
batches.
|
||||
2. Route blocking work (parse, chunk, BM25, `minio`) through
|
||||
`anyio.to_thread.run_sync` with an explicit `CapacityLimiter` created at
|
||||
startup, so ingestion cannot exhaust Starlette's thread pool.
|
||||
3. Enforce the request bounds: `INGESTION_MAX_CONCURRENCY` (`503` + `Retry-After`
|
||||
when exceeded), `INGESTION_TIMEOUT_SECONDS` around the whole work phase
|
||||
(`504`), and the size/chunk-count ceiling (`413`) checked before work starts.
|
||||
4. Guarantee a bounded failure always writes a terminal job status — a timeout
|
||||
must never leave a job stuck in `running`.
|
||||
5. Add unit tests for batching/concurrency limits, timeout-to-terminal-status,
|
||||
and capacity rejection, using scripted embedder fakes.
|
||||
|
||||
**Exit criteria:** an outbox event becomes a durable RabbitMQ message after a
|
||||
publisher restart; retrying publication cannot create duplicate application work.
|
||||
**Exit criteria:** embedding a few hundred chunks issues batched, concurrent
|
||||
requests rather than serial ones; exceeding any bound produces the right status
|
||||
code and a terminal job row.
|
||||
|
||||
### Phase 5: Ingestion worker and Qdrant Chunk/Point CRUD
|
||||
### Phase 5: Ingestion execution and Qdrant Chunk/Point CRUD
|
||||
|
||||
1. Create a separate worker entrypoint with its own application-lifetime database,
|
||||
MinIO, RabbitMQ, Qdrant, model, and logging clients.
|
||||
2. Consume `ingestion.job.created`; reload and validate Postgres records before
|
||||
fetching the MinIO object.
|
||||
3. Transition the job from `queued` to `running` conditionally, append progress
|
||||
events, parse CSV, create deterministic chunks, and upsert tenant-scoped
|
||||
Qdrant points.
|
||||
4. Mark the job `succeeded` with counters or `failed` with a safe error summary;
|
||||
acknowledge only after final/progress state is persisted.
|
||||
5. Make a repeated delivery of the same `ingestion_job_id` safe: no duplicate
|
||||
logical chunks, no incorrect counters, and no transition from a terminal state
|
||||
back to `running`.
|
||||
1. Implement the ingestion service called by the route, using the
|
||||
application-lifetime database, MinIO, Qdrant, model, and logging clients.
|
||||
2. Validate the persisted records before fetching the MinIO object.
|
||||
3. Append progress events, parse CSV, create deterministic chunks, embed them,
|
||||
and upsert tenant-scoped Qdrant points — without holding a Postgres session
|
||||
open across the work.
|
||||
4. In a second short transaction, mark the job `succeeded` with counters or
|
||||
`failed` with a safe error summary, then return the terminal response.
|
||||
5. Make a retried upload safe: no duplicate logical chunks, no incorrect
|
||||
counters, and no transition from a terminal state back to `running`.
|
||||
6. Add unit tests for deterministic CSV chunks, point IDs, and terminal job
|
||||
transitions. Add Testcontainers Qdrant and RabbitMQ integration tests for
|
||||
tenant-filtered upserts, acknowledgement after durable state, redelivery, and
|
||||
transitions. Add Testcontainers Qdrant and Postgres integration tests for
|
||||
tenant-filtered upserts, terminal state persistence, retrying an upload, and
|
||||
parser/Qdrant failure handling.
|
||||
|
||||
**Exit criteria:** a successful upload reaches `succeeded`, and its points are
|
||||
retrievable only under the owning tenant's Qdrant filter. Forced worker failure
|
||||
and message redelivery produce a correct final job state.
|
||||
**Exit criteria:** a successful upload returns `201` with a terminal status, and
|
||||
its points are retrievable only under the owning tenant's Qdrant filter. A forced
|
||||
failure mid-ingestion produces a `failed` job and the right HTTP status, and
|
||||
retrying the upload produces a correct final state without duplicate chunks.
|
||||
|
||||
### Phase 6: Operations, integration tests, and documentation
|
||||
|
||||
1. Add an operator runbook covering local startup, migrations, MinIO bucket setup,
|
||||
RabbitMQ exchange/queue setup, web/worker/outbox commands, and job replay.
|
||||
2. Add a serialized Compose-based operational smoke test where the web process,
|
||||
outbox publisher, and worker run independently for upload through indexed
|
||||
points. Testcontainers remains the standard pytest mechanism for individual
|
||||
adapter integration tests.
|
||||
3. Add end-to-end tests for duplicate upload, duplicate RabbitMQ delivery, outbox
|
||||
publisher crash/restart, worker crash/restart, tenant isolation, and failed
|
||||
parser/Qdrant behavior.
|
||||
1. Add an operator runbook covering local startup, migrations, MinIO bucket
|
||||
setup, the run command, ingestion-bound tuning, the proxy/client timeout
|
||||
requirement, and how to retry a failed ingestion.
|
||||
2. Add a serialized Compose-based operational smoke test covering upload through
|
||||
indexed points against the running web process. Testcontainers remains the
|
||||
standard pytest mechanism for individual adapter integration tests.
|
||||
3. Add end-to-end tests for duplicate upload, retrying a failed upload, tenant
|
||||
isolation, capacity/timeout rejection, and failed parser/Qdrant behavior.
|
||||
4. Add health/readiness checks that distinguish process health from dependency
|
||||
readiness.
|
||||
5. Update the README with local-start instructions and links to ADRs, this plan,
|
||||
@@ -274,10 +285,11 @@ covered by automated tests:
|
||||
```text
|
||||
POST /v1/files (authenticated CSV upload)
|
||||
-> raw bytes stored privately in MinIO
|
||||
-> source file, queued job, and outbox event committed in Postgres
|
||||
-> outbox publisher writes a durable RabbitMQ message
|
||||
-> ingestion worker processes and indexes deterministic Qdrant points
|
||||
-> source file and running job committed in Postgres, connection released
|
||||
-> parse/chunk on threads, embed in bounded concurrent batches
|
||||
-> deterministic Qdrant points upserted
|
||||
-> Postgres records progress and terminal job status
|
||||
-> 201 Created returns the terminal result in the same request
|
||||
-> GET /v1/files/{file_id} reports that status within the owning tenant only
|
||||
```
|
||||
|
||||
|
||||
Reference in New Issue
Block a user