Feature: RAG Ingestion Hardening: Idempotent Uploads, Dedup, Async Indexing, and Progress Events

[hoangsonww]

Summary

Uploads occasionally block while the backend warms up or when document processing (chunking → embedding → indexing) is slow. Let’s harden the RAG pipeline so each upload is idempotent, deduplicated, processed asynchronously via RabbitMQ, and exposes progress events to the UI (and the VS Code extension). Users should never wait on a long synchronous request, and re-uploads of the same file shouldn’t duplicate storage or vectors.

Goals

• Idempotent ingestion keyed by a deterministic doc_fingerprint (content hash).
• Global dedup across users for identical binaries (with per-user ACLs).
• Async indexing job (RabbitMQ) with retries + dead-letter.
• Frontend receives live status via SSE (or WebSocket) to show “Uploading”, “Chunking”, “Embedding”, “Indexed”.
• Safe re-try: repeated /upload returns the existing doc and resumes any pending job.
• Cold-start resilience with short responses + background work.

Non-Goals

• Changing the summarization/chat models.
• Swapping out datastore choices (Mongo/Firestore/Redis) beyond what’s listed below.

---

Design

1) Deterministic Document Identity

• Compute SHA-256 over normalized file bytes (e.g., strip metadata if feasible).

  • Store as doc_fingerprint on the doc record.

• New unique key: (doc_fingerprint, owner_id) to prevent per-user dupes.

• If doc_fingerprint exists globally, link existing parsed text/chunks to the new user (copy-on-write for user fields).

2) Ingestion API (Backend/Express)

New endpoint POST /ingest/init
Returns a short response; creates/returns docId and kicks a job.

Request (multipart or pre-signed URL flow):

POST /ingest/init
Authorization: Bearer <token>
Content-Type: multipart/form-data
file=...

Response:

{ "docId": "doc_123", "status": "queued", "fingerprint": "sha256:..." }

New endpoint GET /ingest/status/:docId

• Emits SSE with { stage, progressPct, message } events.
• Stages: queued | parsing | chunking | embedding | indexing | complete | failed.

3) Async Pipeline (RabbitMQ)

• Producer: ingest/init enqueues { docId, fingerprint }.

• Consumer worker:

  1. Parse (PDF/Docx → text) → store raw text.
  2. Chunk (LangChain) with stable chunk IDs chunk_<fingerprint>_<n>.
  3. Embed & upsert to vector store (idempotent upsert on chunk IDs).
  4. Mark doc indexed=true.

• Retries with exponential backoff; on repeated failures move to DLQ.

• Publish progress to Redis pub/sub → SSE broadcaster.

4) Dedup + Caching

• If fingerprint exists with indexed=true, skip re-embedding; just attach ACL/user link.
• Cache popular doc summaries in Redis: key sum:v1:<fingerprint>:<locale> TTL 24h.
• Cache chat retrieval results for first k turns (short TTL ~5–10m).

5) Frontend/Extension UX

• On upload, immediately show progress stream from /ingest/status/:docId (SSE).
• When complete, fetch summary/ideas endpoints.
• If failed, show retry CTA (fires /ingest/retry/:docId → requeue).

6) Observability

Add Prometheus counters/histograms:

• ingest_jobs_total{stage}
• ingest_failures_total{stage,reason}
• ingest_latency_ms_bucket{stage}
• embedding_upserts_total{dedup="hit|miss"}
  Basic Grafana JSON dashboard in docs/observability/ingest-dashboard.json.

---

Acceptance Criteria

• ✅ Re-uploading the same binary (same user) returns existing docId without duplicate vectors.
• ✅ Upload of a file already indexed by another user completes in < 2s (link + ACL).
• ✅ SSE delivers stage updates; UI shows progress without polling.
• ✅ p95 of /ingest/init < 400 ms (no sync heavy work).
• ✅ On worker crash/restart, job resumes without double-embedding (idempotent chunk IDs).
• ✅ Metrics visible; DLQ contains failed jobs with reason.

---

Tasks

Backend

• Add SHA-256 fingerprinting + (fingerprint, owner_id) unique index (Mongo/Firestore).
• Implement /ingest/init, /ingest/status/:docId (SSE), /ingest/retry/:docId.
• Producer: enqueue job with minimal payload; store status in Mongo.
• Worker: parse → chunk → embed → index; publish progress via Redis → SSE.
• Idempotent vector upsert keyed by chunk_<fingerprint>_<n>.
• Redis caches for summaries & first-turn retrieval.
• Prometheus metrics + health endpoints.

Frontend (React) & VS Code Extension

• Hook upload flow to /ingest/init then open SSE stream.
• Progress UI (“Queued”, “Parsing”, … “Complete”).
• Retry button wired to /ingest/retry/:docId.
• Handle dedup fast-path (show “Already processed — linking to your library”).

DevOps

• RabbitMQ DLQ + dashboards; alerts on failure spikes.
• Add worker container to docker-compose.yml and k8s manifests.
• CI: unit tests for fingerprinting, SSE handler, worker idempotency.

Tests

• Unit: fingerprinting, dedup linking, SSE stream formatting.
• Integration: kill worker mid-run → resume without dupes.
• Load: 50 concurrent uploads; ensure queue drains; p95 latencies met.

---

Risks & Mitigations

• PDF parsing variability → normalize bytes before hashing or store dual keys (raw vs parsed).
• Provider rate limits for embeddings → small batch size + backoff; queue concurrency control.
• SSE behind proxies → keep-alive headers and NGINX proxy_read_timeout config; WebSocket fallback if needed.