Multi-rag storage and read with rollout plan

gistfile1.txt

# Fixing Torn Reads and Silent Doc Loss in a Distributed RAG Pipeline

A design for a distributed RAG system where indexer workers and reader workers
share an object store, and readers intermittently deserialize half-written
embeddings.

## Problem statement

Architecture: an HTTP entrypoint pushes tasks into Redis; a pool of workers pops
those tasks and writes documents and embeddings into a shared store, which the
same or other workers read at query time.

Reported symptoms:

- Readers occasionally load an embedding that is only partially written, so
  deserialization fails.
- That failure escapes to the HTTP layer, and the user receives an error rather
  than an answer.
- Documents that were being written are sometimes never indexed at all.
- Tasks that depended on reading those documents never execute, and nothing
  reports the gap.

Constraints established during design:

| Constraint | Value |
| --- | --- |
| Storage engine | Object storage (S3 / GCS / MinIO) |
| Access path | Multipart / streaming upload |
| Granularity | One object per document |
| Consistency requirement | Stale but valid is acceptable |
| Blast radius | Keep the engine; change the write protocol only |
| Availability | No downtime; the web application stays up throughout |
| Priority | Corruption and durability weighted equally |

## Diagnosis

One symptom, three independent defects. Treating them as a single bug is why the
problem keeps returning.

### A. Torn read

A writer overwrites a live key in place while a reader is fetching it. A
multipart upload to an existing key is atomic on S3 and GCS proper, but not on
every MinIO or Ceph RGW configuration, and never if any code path performs
delete-then-write or an append/compose operation.

### B. Silent loss

`LPOP` / `BRPOP` acknowledges the task at pickup. If the worker dies after
popping and before committing, the task is gone: the document is never indexed
and nothing is notified. This is the "documents get lost" symptom, and it has no
causal relationship to defect A.

### C. Poison propagation

The deserialization error escapes to the HTTP handler and the task dies without
retry. One corrupt document destroys an answer that the other nineteen retrieved
documents could have served perfectly well.

Each defect needs its own mechanism: A is fixed structurally, B with real
acknowledgement semantics, C with graceful degradation.

## Approaches considered

### Option 1 — read-side validation only

Checksum before deserializing, skip bad documents. Cheap and shippable
immediately, and it stops user-facing 500s. It does not stop loss and does not
stop tearing. A useful first deploy, not the fix.

### Option 2 — locking

A per-document lease in Redis: readers take a shared lock, writers take an
exclusive one. Rejected. A lock TTL over an HTTP object store fails open when the
writer crashes, serializes readers, and adds a round trip to every read — while
the store itself still has no idea the lock exists. A distributed lock over a
store that does not enforce it is advice, not safety.

### Option 3 — immutable versions plus pointer commit (recommended)

Multi-version concurrency control implemented on object storage. No locks,
readers never block, and torn reads become structurally impossible rather than
merely rare. Fits the "keep the engine, change the write protocol" constraint
exactly.

## Design

### 1. Write protocol: write once, then flip a pointer

Key layout:

```
emb/{doc_id}/v/{version}/payload      # immutable, never overwritten
emb/{doc_id}/current                  # tiny JSON pointer
```

`version` is a ULID or the sha256 of the content. The writer PUTs the payload to
a key that has never existed before. Multipart upload is now safe on any engine,
because a partially uploaded new key is a key that no reader has heard of yet.

The commit is a single small PUT of the pointer object:

```json
{
  "version": "01J2X8Q4M7ZR6K9F0TB3YH5CWD",
  "sha256": "9f2c...",
  "bytes": 409600,
  "dim": 1536,
  "dtype": "float32",
  "codec": "raw",
  "prev": "01H9V1N4P2QK8M3D7RA5XJ6BFE",
  "written_at": "2026-08-20T16:42:11Z"
}
```

The pointer is small enough that it is never a multipart upload, which makes it
atomic on every engine including the permissive ones. A live payload is never
overwritten again. A reader resolves the pointer, then fetches an immutable key,
and that key cannot change underneath it.

Stale-but-valid is satisfied by construction: a reader arriving mid-flip serves
the previous version.

Enforce the invariant with a bucket policy denying overwrite on the `v/` prefix,
with object versioning enabled, so a future regression fails loudly instead of
silently corrupting data.

### 2. Self-verifying payloads: defense in depth

Wrap the payload in a framed envelope: magic bytes, format version, dtype,
dimension, vector count, the payload itself, and a sha256 trailer. The reader
checks the declared length and the checksum *before* calling any deserialization
routine.

Send `ChecksumSHA256` on the PUT so the store itself rejects a corrupt upload,
and compare the GET `Content-Length` against the `bytes` field in the pointer.

Corruption then reports itself as "document X failed its integrity check" at the
boundary, rather than as an unexplained `frombuffer` error two hundred
milliseconds later in an unrelated module.

### 3. Read path: degrade, never 500

When verification fails, fall back to the `prev` version recorded in the pointer.
If that also fails, drop the document from the candidate set, increment a metric,
enqueue a repair task, and answer from the remaining documents.

RAG is naturally tolerant here: one missing document produces a slightly worse
answer, not an error page.

Add a coverage threshold. If more than N percent of the retrieved documents fail
verification, return an explicit partial-index response instead of a confidently
wrong one. That distinction matters more than the corruption itself — an
incomplete answer labelled as incomplete is recoverable, and a confident wrong
answer is not.

### 4. Queue durability: at-least-once with a reclaimer

Replace the list queue with a Redis Streams consumer group. Use `XADD` to
enqueue, `XREADGROUP` to pick up, and `XACK` **only after the pointer commit
succeeds**. A crash mid-work leaves the entry in the pending-entries list rather
than deleting it.

A reclaimer process runs `XAUTOCLAIM` against a minimum idle time and hands
orphaned entries to a live worker. Once the delivery count passes a threshold,
route the entry to a dead-letter stream and raise an alert. Nothing is ever
dropped silently.

Retries are safe at no extra cost: keys are content-addressed, so a rerun writes
a new version and flips the pointer, and identical content hashes to the same key
and becomes a no-op.

One further rule: **Redis is not the source of truth.** Ingestion writes a
durable intent record — document id mapped to desired state — *before* pushing to
Redis. Total loss of Redis then costs a replay rather than the documents
themselves.

### 5. Reconciliation sweeper

A periodic job diffs desired state (the intent records) against actual state (a
pointer exists and verifies), and re-enqueues the difference. The same pass
garbage-collects versions that are old, unreferenced by any pointer, and past
retention, and a bucket lifecycle rule aborts incomplete multipart uploads.

This is the highest-value component in the design. Every other fix closes a known
failure mode; the sweeper closes the unknown ones, including those introduced by
future changes. A sustained `reconcile_delta` above zero is the alarm for
"documents are silently disappearing" — a condition that currently produces no
signal whatsoever.

### 6. Rollout with the web application live

The ordering is load-bearing. Readers become tolerant before writers change
anything; reversing the order briefly guarantees the outage being fixed.

1. Deploy readers with checksum verification and skip-bad-document degradation,
   leaving all other behavior unchanged. This alone stops user-facing errors on
   day one.
2. Deploy readers that resolve the pointer first and fall back to the legacy key.
   Writers are still untouched.
3. Switch writers to write-once plus pointer commit, optionally dual-writing the
   legacy key during the transition window.
4. Run the sweeper in backfill mode: mint pointers for legacy documents and
   re-embed any that fail verification.
5. Once the backfill is clean, stop legacy writes, drop the reader fallback, and
   enable the deny-overwrite bucket policy.
6. Migrate the queue: new tasks go to Streams while a shim consumer drains the
   old list, which is deleted once empty.

Every step is independently revertible, and no step requires both sides of the
system to deploy simultaneously.

### 7. Observability

Track `verify_fail_total`, `pointer_missing_total`, `stream_pending_depth`,
`dlq_depth`, `reconcile_delta`, and `index_lag_seconds`. Alert on `dlq_depth` and
on a sustained non-zero `reconcile_delta`.

## Summary

- **Write to a new immutable key and commit by flipping a small pointer** — torn
  reads become impossible rather than rare, because readers never observe a key
  that is being written.
- **Deny overwrite at the bucket policy** — a regression fails loudly instead of
  silently corrupting data.
- **Frame payloads with a checksum and length, and verify before deserializing** —
  corruption is caught at the boundary, named correctly, and never becomes a
  mystery stack trace.
- **On verification failure, fall back to the previous version, then drop the
  document** — one bad document degrades an answer instead of returning an error
  page.
- **Return an explicit partial-coverage response when too many documents fail** —
  a confident wrong answer is worse than an honest incomplete one.
- **Use Redis Streams with `XACK` after commit, plus an `XAUTOCLAIM` reclaimer** —
  a worker crash replays the task rather than deleting it.
- **Dead-letter after N attempts** — poison tasks stop retrying but remain
  visible, so nothing vanishes quietly.
- **Write a durable intent record before enqueueing** — Redis becomes a transport
  rather than the system of record.
- **Run a reconciliation sweeper diffing desired against actual state** — it
  catches every gap including future bugs, and converts silently unindexed
  documents into an alertable metric.
- **Order the rollout readers-tolerant first, writers second, policy last** —
  every phase is revertible and the application never goes down.