azk/chatbot_v3
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

docs(architecture): replace NATS JetStream with RabbitMQ for job dispatch

Browse Source 
Switch the durable ingestion/maintenance job-dispatch broker decision from
NATS JetStream to RabbitMQ (aio-pika), rewriting ADR-0014 and propagating
the terminology change through ADR-0015, ADR-0016, and the ingestion
vertical-slice plan. Adds aio-pika as a runtime dependency.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Ali Zarin Kolah
2026-08-15 11:25:25 +03:30
parent 0ca698acfa
commit 88b2db0c3d
6 changed files with 2250 additions and 241 deletions
Download Patch File Download Diff File Expand all files Collapse all files

194
docs/adr/0014-durable-job-dispatch-with-nats-jetstream.md → docs/adr/0014-durable-job-dispatch-with-rabbitmq.md
View File

@@ -1,4 +1,4 @@
# 0014. Durable job dispatch with NATS JetStream
# 0014. Durable job dispatch with RabbitMQ
## Status
@@ -28,13 +28,13 @@ The broker decision has to preserve existing boundaries:
## Decision
Use **NATS JetStream** as the durable pub/sub and job-dispatch broker for
asynchronous ingestion and maintenance work.
Use **RabbitMQ** (via the **aio-pika** async SDK) as the durable pub/sub and
job-dispatch broker for asynchronous ingestion and maintenance work.
NATS core pub/sub alone is not enough for ingestion dispatch because ingestion
jobs need durable delivery, acknowledgement, redelivery, and worker restart
recovery. JetStream provides those broker-level mechanics while Postgres remains
the source of truth for application state.
RabbitMQ's work-queue model gives ingestion dispatch what it needs directly:
durable exchanges/queues, manual acknowledgement, redelivery on nack/crash, and
per-queue dead-lettering for poison messages, all while Postgres remains the
source of truth for application state.
### Start with ingestion jobs
@@ -46,22 +46,26 @@ POST /v1/files
-> store source bytes in MinIO (ADR-0013)
-> create/update source_files row
-> create ingestion_jobs row with status='queued'
-> commit the transaction
-> publish a JetStream message containing ingestion_job_id
-> create an unpublished outbox_events row
-> commit one transaction
-> return 202 Accepted
outbox publisher
-> publish the RabbitMQ message containing durable ids
-> mark the outbox event published
```
A dedicated ingestion worker process consumes the message:
```text
JetStream message
RabbitMQ message
-> load ingestion_jobs/source_files from Postgres
-> bind logging context from durable ids
-> fetch source object from MinIO
-> parse, chunk, embed, and mutate Qdrant points
-> append ingestion_job_events
-> mark ingestion_jobs succeeded/failed/cancelled
-> ack or negatively acknowledge the message
-> ack or nack/reject the message
```
### Worker-owned ingestion chunk CRUD
@@ -69,14 +73,17 @@ JetStream message
For file ingestion and re-ingestion, generated chunk/point CRUD is performed by
the ingestion worker, not by the FastAPI publisher/request path.
The publisher only owns the lightweight durable handoff:
The HTTP request path only owns the lightweight durable handoff:
- authenticate the caller and derive tenant context;
- store uploaded bytes in MinIO;
- create or update `source_files`;
- create `ingestion_jobs(status='queued')`;
- commit the Postgres transaction;
- publish the JetStream message containing durable ids.
- create the corresponding unpublished `outbox_events` row;
- commit the Postgres transaction and return `202 Accepted`.
The separate outbox-publisher process owns publication of the RabbitMQ message
containing durable ids and records its outcome on the outbox event.
The ingestion worker owns the expensive and retryable side effects:
@@ -112,15 +119,15 @@ message is:
```
Do not put raw file bytes, extracted text, chunks, embeddings, raw prompts, model
outputs, or secrets in JetStream messages.
outputs, or secrets in RabbitMQ messages.
Workers must treat message metadata as routing/correlation input, not as the only
authority. Before doing tenant-scoped work, workers reload the durable job and
file rows from Postgres and verify the ids are consistent.
### Use explicit subjects, streams, and durable consumers
### Use explicit routing keys, a topic exchange, and durable queues
Use stable, dot-separated subjects. Initial subjects:
Use stable, dot-separated routing keys. Initial routing keys:
```text
ingestion.job.created
@@ -129,16 +136,21 @@ maintenance.retention.requested
maintenance.erasure.requested
```
Create an application-owned JetStream stream for durable work messages, for
example:
Create an application-owned durable topic exchange for dispatching work
messages, and durable (quorum-type) queues bound to it by pattern, for example:
```text
CHATBOT_JOBS
subjects: ingestion.job.*, maintenance.*
exchange: chatbot.jobs (topic, durable)
queue: ingestion.jobs <- binding pattern ingestion.job.*
queue: maintenance.jobs <- binding pattern maintenance.*
```
Workers use durable consumers or queue groups so multiple worker processes can
share work without processing every message independently.
Multiple worker processes share a queue as competing consumers, so work is
distributed without each process independently receiving every message. Set
consumer prefetch (QoS) explicitly rather than relying on the client default, so
one slow consumer cannot starve the others or accumulate unbounded unacked
messages.
Ack messages only after the worker has persisted the resulting job state and
progress events to Postgres. Redelivery must be safe: ingestion processors should
@@ -146,11 +158,12 @@ be idempotent by `ingestion_job_id` and Qdrant point ids/upsert semantics.
### Run subscribers as worker processes, not FastAPI side effects
FastAPI may create a JetStream publisher/client during application lifespan so
routes can publish job notifications after durable state is committed.
The outbox-publisher process owns the RabbitMQ publisher connection/channel and
publishes committed outbox events. FastAPI routes create durable outbox intent
only; they do not publish ingestion notifications directly.
Subscribers should run in separate worker process entrypoints. Do not start a
subscriber inside every FastAPI web worker by default: with multiple
consumer inside every FastAPI web worker by default: with multiple
Uvicorn/Gunicorn workers, each process has its own lifespan and could create
surprising duplicate consumers or resource pressure.
@@ -159,48 +172,75 @@ Worker processes create their own application-lifetime resources at startup:
- SQLAlchemy engine/sessionmaker;
- MinIO/S3 object-storage client;
- Qdrant client;
- NATS/JetStream client;
- RabbitMQ connection/channel (aio-pika, via `aio_pika.connect_robust` for
automatic reconnection);
- observability/logging clients;
- ingestion model clients.
Each message gets its own request/job-lifetime SQLAlchemy `AsyncSession` and
explicit transaction boundaries, following ADR-0012.
### Handle commit/publish consistency deliberately
### Use a transactional outbox for job dispatch
The initial implementation may publish the JetStream message after the Postgres
transaction commits. This avoids workers observing uncommitted jobs, but it means
a process crash between commit and publish can leave a job in `queued` state with
no broker message.
Use a transactional outbox to atomically record the ingestion job and the intent
to dispatch it. The HTTP publisher does not publish directly to RabbitMQ after
committing the ingestion-job transaction.
To handle that failure mode, workers or a small scheduler should periodically
scan Postgres for queued/stale ingestion jobs and publish or process them again.
This keeps Postgres authoritative and makes JetStream a wakeup/delivery
mechanism rather than the only source of pending work.
If stricter guarantees become necessary, introduce a transactional outbox table
in Postgres via a future ADR/migration:
Instead, the same Postgres transaction creates or updates the durable job state
and inserts an unpublished outbox event:
```text
same DB transaction:
insert/update ingestion_jobs
insert outbox_events
publisher process:
read unpublished outbox_events
publish to JetStream
mark outbox event published
insert/update source_files
insert ingestion_jobs(status='queued')
insert outbox_events(type='ingestion.job.created', payload, published_at=NULL)
commit
```
Do not add DDL at FastAPI startup; all outbox schema changes must go through
Alembic as required by ADR-0009.
A separate outbox publisher process reads unpublished events, publishes the
versioned message to RabbitMQ, and records the publication outcome:
```text
outbox publisher:
claim unpublished outbox event
publish to RabbitMQ (persistent message, publisher confirms enabled)
mark outbox event published once the broker confirms receipt
```
This removes the failure window where an `ingestion_jobs` row commits but the API
process crashes before recording that it must be dispatched. If the transaction
commits, the job and its durable dispatch intent both exist; the outbox publisher
can resume publication after a restart.
An outbox event must have a stable UUID event id. Include that id in the
RabbitMQ message so consumers and operators can correlate it back to the outbox
record. RabbitMQ has no broker-native publish-deduplication mechanism comparable
to some other brokers: publisher confirms guarantee the broker accepted the
message, but the publisher can still crash after the broker confirms and before
`published_at` is recorded, causing a duplicate publish on retry. There is no
broker-side safety net for that case here — **workers must be idempotent by
`ingestion_job_id`**, and Qdrant mutations must use deterministic point
ids/upsert semantics, as the sole guarantee against duplicate processing.
Outbox events are delivery records, not a replacement for `ingestion_jobs` or
`ingestion_job_events`. Postgres continues to own application job state and
progress. A stale-job scanner remains a repair/operational check for jobs or
outbox events that have not advanced as expected, not the primary dispatch path.
Add the `outbox_events` schema through an Alembic migration. Do not create or
alter the table at FastAPI startup, as required by ADR-0009.
### Configure retries, dead letters, and observability
JetStream consumers should configure explicit acknowledgement, redelivery, and
maximum delivery behavior. Poison messages or repeatedly failing jobs should end
as durable Postgres failures with an `ingestion_job_events` error entry, not an
infinite invisible retry loop.
Declare queues with an explicit dead-letter exchange (`x-dead-letter-exchange`)
pointing at a durable `chatbot.jobs.dlx` fanout exchange bound to a durable
dead-letter queue. A message lands there when a worker rejects/nacks it with
`requeue=False`. Workers should nack with `requeue=False` after a bounded local
retry count for a given delivery, rather than looping redelivery indefinitely.
Poison messages or repeatedly failing jobs should end as durable Postgres
failures with an `ingestion_job_events` error entry, and the dead-letter queue
gives operators a place to inspect the raw message for messages that failed
before any Postgres state could be written.
At worker ingress, bind structured logging context from durable identifiers:
@@ -209,7 +249,7 @@ At worker ingress, bind structured logging context from durable identifiers:
- `api_key_id`;
- `file_id`;
- `ingestion_job_id`;
- JetStream stream/subject/sequence when available.
- RabbitMQ exchange/routing key/delivery tag when available.
Use stable event names such as:
@@ -220,7 +260,7 @@ Use stable event names such as:
- `broker.publish.failed`;
- `broker.message.redelivered`.
### Do not queue chat runs through JetStream yet
### Do not queue chat runs through RabbitMQ yet
ADR-0007 and ADR-0008 intentionally reject queueing concurrent chat runs per
thread: one run may be in flight, and concurrent runs return `409 Conflict`.
@@ -235,45 +275,55 @@ execution remains owned by the FastAPI/LangGraph boundary unless superseded.
### Positive
- File ingestion can move out of HTTP request handling into scalable worker
processes without changing the REST contract from ADR-0008.
- JetStream gives durable delivery, ack/redelivery, and worker-group semantics
without adopting Kafka-level operational complexity.
- RabbitMQ gives durable delivery, manual ack/redelivery, per-queue
dead-lettering, and competing-consumer work distribution without adopting
Kafka-level operational complexity.
- Postgres remains the auditable source of truth for jobs, progress, tenant
ownership, and failure state.
ownership, failure state, and the durable intent to dispatch each job.
- The transactional outbox prevents a committed job from losing its initial
dispatch intent when the API process fails before a direct broker publish.
- MinIO, Postgres, Qdrant, and the broker each have distinct responsibilities:
bytes, metadata/state, vector search, and work delivery.
- Separate worker processes avoid surprising subscriptions in every FastAPI web
worker and make GPU/model-heavy ingestion resources easier to size.
### Negative
- Adds another external service to run, secure, monitor, and back up where stream
durability is required.
- Adds another external service to run, secure, monitor, and back up where
queue/exchange durability is required.
- Developers must reason about redelivery and idempotency; handlers may run more
than once for the same `ingestion_job_id`.
- Publishing after commit has a crash window unless a stale-job scanner or future
transactional outbox is implemented.
- NATS/JetStream stream, consumer, retention, and dead-letter configuration must
- Adds an `outbox_events` table, Alembic migration, outbox publisher process, and
monitoring for unpublished or stuck events.
- RabbitMQ has no broker-native publish deduplication. Unlike brokers that offer
one, duplicate publication after a publisher crash between broker confirmation
and outbox-event marking is not mitigated by the broker at all; idempotent
consumers are the only protection.
- RabbitMQ exchange, queue, binding, and dead-letter-exchange configuration must
be managed explicitly per environment.
- The app now needs worker deployment/runbook conventions in addition to the
FastAPI web process.
- The app now needs web, ingestion-worker, and outbox-publisher deployment/runbook
conventions.
## Alternatives Considered
- **NATS core pub/sub only**: rejected. Core pub/sub is useful for ephemeral
events, but ingestion dispatch needs durable delivery and acknowledgement.
JetStream is the NATS feature that provides those semantics.
- **NATS JetStream**: rejected. JetStream keeps the broker footprint lighter and
offers built-in publish deduplication, but the project wants RabbitMQ's more
mature work-queue/routing model (topic exchanges, per-queue DLX, prefetch-based
fair dispatch) and the aio-pika SDK for asyncio integration.
- **Redis Streams**: rejected for this decision. Redis Streams would work for an
MVP and has simple local deployment, but NATS JetStream is a cleaner dedicated
broker for durable pub/sub and worker coordination without also mixing cache
MVP and has simple local deployment, but RabbitMQ is a cleaner dedicated broker
for durable pub/sub and worker coordination without also mixing cache
responsibilities into the same service.
- **RabbitMQ**: rejected for now. It is mature for work queues and routing, but
the project does not currently need its exchange/routing complexity, and NATS
JetStream keeps the broker footprint lighter.
- **Kafka**: rejected. Kafka is an excellent durable event log at high scale, but
its operational overhead is unnecessary for this service's initial ingestion
and maintenance jobs.
- **FastAPI `BackgroundTasks` or in-process asyncio tasks**: rejected as the
durable design. They are simple, but work can be lost on process restart and
they do not coordinate multiple workers cleanly.
- **Post-commit direct RabbitMQ publish plus a repair scanner**: rejected as the
primary dispatch design. It is simpler, but a crash after the job transaction
commits and before the publish is recorded can strand a queued job until the
scanner finds it. The transactional outbox persists the dispatch intent in the
same transaction as the job.
- **Database polling only**: rejected as the main dispatch mechanism. Polling
Postgres for queued jobs is a useful repair path, but relying on polling alone
adds latency and unnecessary database load once a broker is available.

277
docs/adr/0015-modular-monolith-package-architecture.md Normal file
View File

@@ -0,0 +1,277 @@
# 0015. Modular monolith package architecture
## Status
Proposed
## Context
The repository currently contains only a small FastAPI-oriented scaffold under
`src/`, while the ADRs define several substantial capabilities and external
boundaries: FastAPI HTTP endpoints, Postgres and Alembic, MinIO object storage,
RabbitMQ with a transactional outbox, Qdrant point/retrieval operations,
LangGraph conversational execution, and separate worker processes.
Without an explicit package structure, implementation can drift toward route
handlers that call SDK clients directly, workers that duplicate HTTP logic, and
generic catch-all directories such as `utils`, `services`, or `clients`. That
would make tenant isolation, transactions, retries, resource ownership, and tests
harder to apply consistently.
ADR-0012 requires application-lifetime resource ownership and explicit dependency
passing. ADR-0014 requires workers to own ingestion-generated Chunk/Point CRUD,
while direct point routes remain synchronous API work. The repository structure
must support both entrypoints reusing the same application behavior without
coupling worker code to FastAPI routes or LangGraph nodes to SDK details.
## Decision
Use a **modular monolith with explicit infrastructure adapters**. The application
uses one deployable codebase with separate web, worker, and outbox-publisher
process entrypoints. Business/application behavior is grouped by capability;
external systems are isolated behind infrastructure adapters.
Use `src/` as the explicit Python package root.
### Package layout
Create packages as they become necessary, following this structure:
```text
src/
├── __init__.py
├── main.py
├── config.py
├── bootstrap/
│ ├── lifespan.py
│ └── dependencies.py
├── api/
│ ├── dependencies/
│ ├── routers/
│ ├── schemas/
│ └── router.py
├── application/
│ ├── files/
│ ├── ingestion/
│ ├── points/
│ ├── retrieval/
│ ├── threads/
│ └── ports/
├── agent/
│ ├── graph.py
│ ├── state.py
│ ├── nodes/
│ ├── prompts/
│ ├── tools/
│ └── persistence.py
├── infrastructure/
│ ├── postgres/
│ │ ├── models/
│ │ ├── repositories/
│ │ ├── database.py
│ │ └── outbox.py
│ ├── qdrant/
│ ├── minio/
│ ├── rabbitmq/
│ ├── langgraph/
│ └── observability/
├── messaging/
│ ├── events.py
│ ├── subjects.py
│ └── outbox_publisher.py
└── workers/
├── ingestion.py
└── maintenance.py
```
Keep Alembic configuration and migration revisions at the repository root:
```text
alembic.ini
alembic/
├── env.py
└── versions/
```
Keep tests outside the application package and organize them by testing boundary:
```text
tests/
├── unit/
│ ├── application/
│ └── agent/
├── integration/
│ ├── postgres/
│ ├── minio/
│ ├── rabbitmq/
│ └── qdrant/
└── e2e/
```
### Dependency direction
Entry adapters call application services; application services depend on typed
ports/contracts; infrastructure packages implement those ports.
```text
FastAPI routes / RabbitMQ workers / LangGraph nodes
-> application services
-> application ports
-> infrastructure adapters
```
The reverse direction is prohibited:
- infrastructure adapters do not import FastAPI routers, workers, or graph nodes;
- workers do not call FastAPI route functions;
- LangGraph nodes do not call API route functions or construct mutable SDK clients;
- application services do not import concrete MinIO, RabbitMQ, Qdrant, or
SQLAlchemy client construction code;
- routes do not call raw Qdrant, MinIO, or RabbitMQ SDK methods directly.
Use ports selectively for external side effects and persistence boundaries; do not
add interfaces around pure local functions merely to satisfy a pattern.
### API package
`api/` is the HTTP adapter only:
- `routers/` map HTTP operations to application-service calls;
- `dependencies/` resolve request-lifetime objects such as `AuthContext` and
`AsyncSession`;
- `schemas/` contains public Pydantic request/response/error models;
- `router.py` composes versioned route groups.
API schemas are separate from SQLAlchemy ORM models and RabbitMQ message schemas.
Routes perform HTTP validation and response mapping, but not parsing, embedding,
Qdrant mutations, or transaction-independent business workflows.
### Application package
`application/` contains reusable use-case behavior. It has no FastAPI request
objects, RabbitMQ consumer loops, or SDK client construction.
- `files/` creates source-file records, validates lifecycle actions, and returns
file/job status.
- `ingestion/` performs parse/chunk/embed/index orchestration after the worker
receives a job.
- `points/` applies tenant-aware direct Point CRUD rules and shared generated-point
mutation behavior.
- `retrieval/` owns retrieval use cases used by the graph; it does not expose raw
Qdrant SDK details.
- `threads/` coordinates run-level application behavior without taking ownership
of conversation/session records reserved for the main backend and LangGraph.
- `ports/` defines narrow contracts for external side effects, including object
storage, message publishing, point storage, and repositories where useful.
Both FastAPI routes and worker consumers call these services. This prevents a
second, inconsistent ingestion implementation from growing inside `workers/`.
### Agent package
All LangGraph-specific application graph code lives in `agent/`:
- `graph.py` builds and compiles the graph from explicitly passed dependencies;
- `state.py` defines graph state and graph-facing result types;
- `nodes/` contains focused graph-node behavior such as triage, retrieval,
generation, verification, and memory extraction;
- `prompts/` holds prompt identifiers/templates or prompt access helpers;
- `tools/` contains graph tool definitions;
- `persistence.py` contains graph-facing persistence configuration/types.
Concrete `AsyncPostgresSaver` and `AsyncPostgresStore` setup belongs in
`infrastructure/langgraph/`, then is passed into the graph factory during
bootstrap. Graph nodes call application services, particularly
`application/retrieval/`, instead of embedding Qdrant query logic.
### Infrastructure package
`infrastructure/` contains concrete integrations and resource setup.
- `postgres/` owns SQLAlchemy engine/sessionmaker setup, ORM models, repository
implementations, and transactional outbox persistence.
- `qdrant/` owns Qdrant client lifecycle, collection/bootstrap helpers, low-level
point operations, and hybrid retrieval adapter mechanics.
- `minio/` implements object-storage operations against MinIO/S3-compatible APIs.
- `rabbitmq/` owns the RabbitMQ connection/channel lifecycle plus low-level
publish and consumer adapters (aio-pika).
- `langgraph/` configures the concrete Postgres-backed LangGraph persistence
adapters.
- `observability/` configures structlog and Langfuse integrations.
Infrastructure code receives configuration and is created by a process owner; it
must not create mutable external clients at import time.
### Messaging and workers
`messaging/` contains versioned event schemas, stable routing-key names, and the
outbox-publisher orchestration. The outbox publisher coordinates Postgres outbox
records with the RabbitMQ adapter; it does not become a second source of job
state.
`workers/` contains thin process entrypoints and consumer loops. A worker creates
application-lifetime dependencies, consumes a durable RabbitMQ message, binds
job logging context, and invokes the corresponding application service. It does
not hold parsing/chunking/Qdrant business logic itself.
For example:
```text
workers/ingestion.py
-> application/ingestion/processor.py
-> application/points/service.py
-> infrastructure/qdrant/points.py
```
### Bootstrap and resource ownership
`bootstrap/` composes configuration and concrete infrastructure adapters for each
process entrypoint. FastAPI lifespan owns web-process resources; worker and outbox
publisher startup own their corresponding resources. This implements ADR-0012
without turning `app.state` or module globals into an untyped service locator.
## Consequences
### Positive
- The repository has clear homes for LangGraph, Qdrant, MinIO, RabbitMQ, Postgres,
API, worker, and outbox-publisher code before implementation grows.
- HTTP routes, worker consumers, and LangGraph nodes reuse application services
while remaining separate transport/execution adapters.
- SDK-specific details are isolated, making integration tests and test doubles
practical without hiding all code behind unnecessary abstractions.
- The layout directly supports ADR-0012's explicit resource ownership and
ADR-0014's worker-owned ingestion Chunk/Point CRUD.
- A single codebase remains simple to deploy while allowing web, worker, and
outbox-publisher processes to scale independently.
### Negative
- The initial directory structure is more elaborate than a route-plus-models
FastAPI starter application.
- Developers must maintain dependency direction rather than importing a concrete
client wherever it is convenient.
- Some capabilities span several packages by design, for example an upload route,
application service, MinIO adapter, outbox repository, and publisher process.
- Ports/contracts should remain narrow; excessive abstraction would add ceremony
without improving testability or substitutability.
## Alternatives Considered
- **Technology-first packages only**: rejected. Directories such as `db`,
`qdrant`, `rabbitmq`, and `langgraph` are immediately discoverable, but feature
workflows become scattered across every integration package and encourage
transport adapters to own business behavior.
- **Feature-first packages only**: rejected. Keeping all file, point, and thread
code together is attractive, but it obscures ownership of shared external
clients and risks duplicating infrastructure integration logic across features.
- **Full clean architecture with interfaces for every class/function**: rejected.
The application needs clear external boundaries, not abstraction around pure
helper functions. Ports are reserved for persistence and external side effects.
- **Microservices for ingestion, retrieval, and chat from the start**: rejected.
The project needs independent web/worker processes, but a single modular
codebase avoids premature network boundaries, deployment complexity, and
distributed transaction concerns.
- **Put LangGraph under `api/` or workers under `ingestion/` only**: rejected.
LangGraph and RabbitMQ workers are independent execution adapters; putting one
under another would invert dependencies and make reuse/testing harder.

215
docs/adr/0016-testing-strategy-and-quality-gates.md Normal file
View File

@@ -0,0 +1,215 @@
# 0016. Testing strategy and quality gates
## Status
Proposed
## Context
The project has ADRs for tenant-scoped ingestion, explicit resource ownership,
MinIO object storage, transactional outbox dispatch, RabbitMQ workers,
Qdrant indexing, and a modular monolith. It has no test runner, test fixtures,
or executable test suite yet.
The first CSV ingestion slice has correctness properties that cannot be left to
manual testing: Alembic is the only schema-management path; tenant identity is
trusted server-side context; a file, job, and outbox event commit atomically;
workers are safe under at-least-once delivery; and generated Qdrant points are
idempotent and tenant-filtered. ADR-0015 already reserves a test layout by
boundary, while ADR-0012 requires explicit dependencies and resource lifetimes
that should make tests practical without import-time client patching.
Tests need to give fast feedback during implementation without replacing
integration coverage with mocks or making routine development depend on Docker,
provider credentials, live models, or Langfuse availability.
ADR-0014's transactional-outbox decision controls ingestion dispatch. The upload
path records durable dispatch intent; a separate outbox publisher publishes it.
## Decision
### Use pytest with explicit async support
Use pytest as the test runner. Add `pytest`, `pytest-asyncio`, `httpx`,
`asgi-lifespan`, `testcontainers`, `pytest-timeout`, and `pytest-cov` as
development dependencies.
Configure `pytest-asyncio` in strict mode. Async tests and fixtures must be
explicit. FastAPI tests use `httpx.AsyncClient`, `ASGITransport`, and
`LifespanManager` so they exercise application startup and shutdown according to
ADR-0012. `FastAPI.TestClient` is not the default project test client.
Register these markers:
- one primary boundary marker per test: `unit`, `integration`, or `e2e`;
- `postgres`, `minio`, `rabbitmq`, or `qdrant` for the real service used by an
integration test;
- `slow` only where a test materially exceeds the normal integration feedback
target;
- `live_provider` for an opt-in, credential-gated external-provider smoke test.
Name tests `test_<unit>_<scenario>_<outcome>` and use Arrange–Act–Assert.
### Test by architectural boundary
Use the test layout reserved by ADR-0015, with shared support for fixtures and
assertions:
```text
tests/
├── conftest.py
├── fakes.py
├── support/
│ ├── factories.py
│ └── assertions.py
├── unit/
│ ├── application/
│ └── agent/
├── integration/
│ ├── postgres/
│ ├── minio/
│ ├── rabbitmq/
│ └── qdrant/
└── e2e/
```
- **Unit tests** are the default development feedback loop. They cover pure
application policy, validation, lifecycle transitions, tenant propagation,
deterministic IDs, CSV chunking, event construction, and LangGraph
control-flow policy.
- **Integration tests** validate a production infrastructure adapter against its
real backing service and its deployment-relevant behavior.
- **End-to-end tests** prove a small vertical-slice acceptance contract through
the real application composition. They do not replace faster unit or adapter
tests.
Hand-written fakes and spies implement narrow application-owned ports, not
MinIO, RabbitMQ, Qdrant, or model SDK-shaped interfaces. Scripted model, embedder,
retrieval, clock, and UUID fakes make normal test runs deterministic.
### Apply pragmatic TDD
For application behavior, HTTP contracts, database migrations, reliability
rules, and defects, first write a focused failing test that describes the
observable requirement. Make the smallest change that passes it, then refactor
while the relevant suite is green.
TDD applies to behavior and regressions, not as an artificial ritual for pure
refactors or configuration-only changes with no observable behavior change.
Those changes must preserve and extend existing relevant coverage as needed.
Before introducing a concrete adapter, write tests for the consuming
application port. Before each Alembic migration, write the empty-database
migration test or extend the existing migration test. Add the corresponding
real-adapter integration test before declaring that boundary complete.
### Use disposable real infrastructure in integration tests
Use Testcontainers as the standard automated integration-test resource mechanism
for Postgres, MinIO, RabbitMQ, and Qdrant.
- Tests never connect to a developer's local services or Langfuse-owned storage
and credentials.
- Start containers at suite or session scope, then isolate data per test with
unique data, object-key prefixes, exchange/queue/binding names, and collection
names.
- Disable parallel integration execution until fixture isolation and cleanup are
proven worker-safe.
- Fixtures expose typed settings or connection values. Tests create production
adapters through their normal constructors.
Docker Compose remains the mechanism for manual local validation and a later,
serialized operational smoke test where web, outbox-publisher, and worker run as
independent processes. It is not the default pytest fixture mechanism.
### Treat invariants as reusable contracts
Test the following requirements at the applicable application, adapter, and E2E
boundaries:
- Tenant identity comes from server-side authenticated context. Request payloads,
query parameters, object metadata, and broker messages cannot override it.
- Cross-tenant access does not disclose tenant-owned data. Public routes normally
return `404` for inaccessible resources.
- Alembic creates the schema from an empty database. Tests never use
`Base.metadata.create_all()`, and FastAPI startup performs readiness checks only,
never DDL.
- The upload transaction records `source_files`, a queued `ingestion_jobs` row,
and an unpublished `outbox_events` row atomically. The HTTP route does not
directly publish the ingestion event.
- Broker messages contain durable identifiers and correlation metadata only. The
worker reloads job and source-file records from Postgres before tenant-scoped
side effects.
- Worker acknowledgement follows durable progress or terminal-state persistence.
Duplicate publication and redelivery do not regress terminal jobs, inflate
counters, or create duplicate logical chunks.
- MinIO keys are server-derived internal paths. Qdrant reads and mutations use a
server-derived tenant filter, deterministic point IDs, and upsert semantics.
### Keep correctness tests separate from model-quality evaluation
Pytest verifies deterministic application behavior: graph policy, schemas,
redaction, correlation metadata, retry budgets, and graceful observability
failure handling. Normal pytest runs never call a paid or live model provider.
Langfuse datasets and experiments evaluate prompts, model behavior, retrieval,
citations, and response quality. They are promotion evidence, not a replacement
for application correctness tests. Live-provider smoke tests, if introduced,
are opt-in, credential-gated, rate-limited, and excluded from ordinary local and
pull-request runs.
### Establish phased quality gates
Initially require Ruff format checking, Ruff linting, Ty type checking, and unit
tests. Require Docker-capable integration tests as their adapters are
implemented. Run the separate-process Compose E2E smoke as a serialized
pre-release or scheduled gate until it is reliable enough for every pull request.
Collect coverage reports but do not set a percentage threshold before the first
vertical slice has meaningful implementation. Later introduce a scoped,
ratcheting threshold rather than encouraging low-value coverage.
## Consequences
### Positive
- Unit tests provide fast, deterministic TDD feedback for core application
behavior.
- Real-service tests cover the behaviors least safe to simulate: Alembic
migrations, object storage, RabbitMQ acknowledgements/redelivery, and Qdrant
filtering/upserts.
- Explicit fakes reinforce the dependency direction and resource ownership rules
from ADR-0012 and ADR-0015.
- The ingestion path has concrete tenant-isolation and reliability contracts,
rather than only a happy-path demonstration.
- Live model quality can improve through Langfuse experiments without making
application tests nondeterministic or expensive.
### Negative
- Docker is required for integration and E2E suites.
- Testcontainers add setup time and require careful fixture cleanup.
- Maintaining real-service coverage and test data isolation adds engineering
effort.
- E2E tests do not prove semantic quality of LLM responses; that remains an
evaluation responsibility.
## Alternatives Considered
- **Mock all external SDKs**: rejected. Mocks cannot prove migrations, real
RabbitMQ acknowledgement/redelivery behavior, MinIO semantics, or Qdrant
tenant filtering.
- **Use full-stack Compose tests only**: rejected. They are slow and opaque for
the default development loop and make failures difficult to localize.
- **Run all integration containers on every pytest invocation**: rejected. Test
boundaries should be selected deliberately for fast local feedback.
- **Use live model providers in routine tests**: rejected because they are
nondeterministic, costly, slow, credential-dependent, and hard to assert.
- **Use Langfuse as the primary regression-test runner**: rejected. Langfuse is
the quality/evaluation plane; pytest remains the deterministic application test
framework.
- **Create schemas with `Base.metadata.create_all()` in fixtures**: rejected. It
bypasses the production Alembic migration path.
- **Require strict test-first work for every non-behavioral refactor**: rejected.
TDD should protect observable behavior and regressions, not add ceremony where
no behavior changes.

287
docs/plans/001-ingestion-vertical-slice.md Normal file
View File

@@ -0,0 +1,287 @@
# 001. Ingestion vertical-slice implementation plan
## Purpose
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.
This is an implementation plan, not an Architecture Decision Record. ADRs explain
why major technologies and boundaries were chosen; this document defines the
order, scope, and verification criteria for implementing them.
## Architecture baseline
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. |
| 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. |
| Qdrant | Tenant-filtered generated chunks and their vectors/payloads. |
The controlling ADRs are:
- [ADR-0008](../adr/0008-rest-api-and-fastapi-boundary.md): FastAPI REST boundary
and job-shaped file ingestion contract.
- [ADR-0009](../adr/0009-postgres-sqlalchemy-alembic-schema.md): Postgres source
files, jobs, audit, and migration conventions.
- [ADR-0012](../adr/0012-application-resource-lifetime-and-dependency-ownership.md):
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.
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
them.
## First release scope
### In scope
- `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.
- 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.
- Automated tests for the critical state transitions, redelivery, and tenant
boundaries.
### Explicitly out of scope
- XLSX, DOCX, and legacy DOC ingestion.
- The conversational LangGraph API and SSE streaming.
- Final reranker selection, GPU deployment, or unresolved model licensing from
ADR-0005.
- Direct `/v1/points` CRUD endpoints beyond the reusable service layer required by
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.
## 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.
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
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.
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.
## Required decisions before implementing the affected phase
The first implementation should use these defaults unless a later ADR changes
them:
### Source-file idempotency and replacement
- Use `(tenant_id, domain, content_sha256)` to recognize identical uploads.
- An identical active upload should return the existing source-file/job reference
rather than create a duplicate ingestion.
- A changed upload creates a new ingestion job. Existing active Qdrant points are
replaced only after the new job completes successfully, so a failed re-ingestion
does not remove a working index.
- Preserve the original filename in Postgres metadata. MinIO object keys remain
internal ID-based paths.
This policy should be made explicit in ADR-0009 before implementing re-ingestion
rather than becoming an accidental repository behavior.
### File deletion
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;
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
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.
These are deployment/runbook settings, not new ADRs unless they change the
reliability guarantee or system boundary.
## Build order
### Phase 1: Foundation and local dependencies
1. Add typed configuration in `src/config.py` for Postgres, MinIO, RabbitMQ,
Qdrant, 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.
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.
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.
6. Create FastAPI lifespan setup and typed dependency helpers without creating
schema at startup.
**Exit criteria:** local infrastructure starts; readiness checks can report each
required dependency; clients are opened/closed by process owners; fast unit tests
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`.
3. Implement explicit repositories/services with a request/job-lifetime
`AsyncSession`; routes/services own commit/rollback boundaries as specified by
ADR-0012.
4. Write the empty-database migration test before each schema revision. Create
Testcontainers-based Postgres fixtures for tenants, hashed API keys, database
sessions, and migrations. Do not use `create_all()` in test fixtures.
**Exit criteria:** migrations create the schema from an empty database; application
startup performs no DDL; schema and repository tests verify tenant-scoped
reads/writes and valid job transitions.
### Phase 3: MinIO upload and durable job creation
1. Implement API-key authentication and `AuthContext` tenant derivation.
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.
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
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.
### Phase 4: RabbitMQ and outbox publisher
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.
**Exit criteria:** an outbox event becomes a durable RabbitMQ message after a
publisher restart; retrying publication cannot create duplicate application work.
### Phase 5: Ingestion worker 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`.
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
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.
### 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.
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,
and the operations runbook.
**Exit criteria:** a new developer can start the stack, apply migrations, upload a
CSV, observe the job through completion, and understand how to investigate or
retry a failure.
## Definition of done for the vertical slice
The first slice is done when the following path works in local Compose and is
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
-> Postgres records progress and terminal job status
-> GET /v1/files/{file_id} reports that status within the owning tenant only
```
The next implementation work after this slice is direct Point CRUD, retrieval,
and then the LangGraph conversational flow. Do not couple those later milestones
to the initial ingestion path unless they are needed to preserve one of the
invariants above.