Implementing Read-After-Write Consistency in RAG: Synchronizing Document Ingestion, Embedding, and Index Availability for Real-Time Knowledge Updates

Introduction

Imagine you've built a support chatbot for a SaaS product. A critical security patch ships at 3:47 PM. Your technical writer publishes the updated runbook to your knowledge base. A customer asks about the patch at 3:52 PM — and your RAG system confidently serves stale documentation, missing the fix entirely. The runbook exists in your database. It's just not in your vector index yet.

This is the read-after-write consistency problem, and it's one of the most underappreciated failure modes in production RAG systems. Most tutorials demonstrate ingestion and querying as two separate offline operations. In the real world, documents arrive continuously, and users query immediately after upload. The gap between "document stored" and "document retrievable" can range from seconds to hours depending on how your pipeline is architected — and that gap has real consequences.

By the end of this lesson, you'll understand exactly how to close that gap. We'll design and implement a synchronous ingestion pipeline with built-in consistency guarantees, build async coordination mechanisms for high-throughput scenarios, and instrument the whole thing so you know precisely when documents become queryable.

What you'll learn:

• Why read-after-write inconsistency happens in RAG pipelines and how to diagnose it
• How to design a synchronous ingestion-to-index pipeline with explicit readiness signaling
• How to implement async ingestion with polling, webhooks, and promise-based coordination
• How to handle partial failures mid-pipeline (document stored but embedding failed)
• How to test consistency guarantees under concurrent load

---

Prerequisites

You should be comfortable with the fundamentals of RAG pipelines — specifically the full ingestion lifecycle including document loading, cleaning, and chunking. You should also understand how vector databases store and retrieve embeddings at a production level, and have basic familiarity with async Python (asyncio, aiohttp). SQL literacy is helpful for the state tracking examples.

---

The Anatomy of Inconsistency

Before fixing the problem, let's be precise about where it lives. A typical RAG ingestion pipeline moves a document through several distinct stages:

1. Ingest — Document accepted and stored in raw form (blob storage, database)
2. Parse — Text extracted from PDF/DOCX/HTML
3. Chunk — Text split into retrieval-sized pieces
4. Embed — Each chunk converted to a vector via an embedding model
5. Index — Vectors written to the vector store and become queryable

Each stage transition is a potential gap. If your API endpoint returns 200 OK after Stage 1 (the document is stored) but your user queries the system before Stage 5 completes, they're querying a universe that doesn't include their document yet.

The gap is worse than it looks because embedding is typically the bottleneck. For a 50-page technical PDF:

• Storage: ~50ms
• Parsing: ~300ms
• Chunking: ~100ms
• Embedding (via OpenAI API, 200 chunks): ~8-15 seconds
• Vector index write: ~200ms

That's potentially 15+ seconds of inconsistency at minimum, and that assumes your pipeline is processing synchronously in the foreground. If you're using an async queue (Celery, SQS, a cron job), the window could be minutes.

---

Designing a State Machine for Document Ingestion

The foundation of read-after-write consistency is explicit state tracking. Your pipeline needs a canonical, queryable record of where every document is in its lifecycle. Think of it as a state machine with the following states:

RECEIVED → PARSING → CHUNKING → EMBEDDING → INDEXING → READY
                                                        ↓
                                               FAILED (at any stage)

Here's a PostgreSQL table that models this cleanly:

CREATE TABLE document_ingestion_jobs (
    job_id          UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    document_id     UUID NOT NULL,
    tenant_id       UUID NOT NULL,
    filename        TEXT NOT NULL,
    stage           TEXT NOT NULL DEFAULT 'RECEIVED',
    -- RECEIVED | PARSING | CHUNKING | EMBEDDING | INDEXING | READY | FAILED
    chunk_count     INTEGER,
    embedded_chunks INTEGER DEFAULT 0,
    indexed_chunks  INTEGER DEFAULT 0,
    error_stage     TEXT,
    error_message   TEXT,
    created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    ready_at        TIMESTAMPTZ
);

CREATE INDEX idx_ingestion_jobs_document ON document_ingestion_jobs(document_id);
CREATE INDEX idx_ingestion_jobs_tenant_stage ON document_ingestion_jobs(tenant_id, stage);

This table becomes the source of truth. Every stage transition updates a row. Your query endpoint can join against this table to filter out documents that haven't reached READY yet — or return a meaningful status to callers who ask.

---

Implementing Synchronous Ingestion with Explicit Readiness

For many use cases — especially internal tooling where document volumes are moderate — the simplest solution is the right one: run the entire pipeline synchronously and don't return success until the document is actually queryable.

Here's a full Python implementation using LangChain's document loaders, OpenAI embeddings, and pgvector:

import asyncio
import uuid
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional

import asyncpg
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import PyPDFLoader
from langchain_openai import OpenAIEmbeddings
from pgvector.asyncpg import register_vector

class IngestionPipeline:
    def __init__(self, db_pool: asyncpg.Pool, embedder: OpenAIEmbeddings):
        self.db = db_pool
        self.embedder = embedder
        self.splitter = RecursiveCharacterTextSplitter(
            chunk_size=512,
            chunk_overlap=64,
            separators=["\n\n", "\n", ". ", " ", ""]
        )

    async def ingest_document(
        self,
        file_path: str,
        tenant_id: str,
        document_id: Optional[str] = None,
    ) -> dict:
        """
        Fully synchronous ingestion. Returns only when document is READY.
        Raises on any pipeline failure with stage context.
        """
        document_id = document_id or str(uuid.uuid4())
        job_id = str(uuid.uuid4())
        filename = Path(file_path).name

        # Create the job record immediately
        async with self.db.acquire() as conn:
            await conn.execute(
                """
                INSERT INTO document_ingestion_jobs
                    (job_id, document_id, tenant_id, filename, stage)
                VALUES ($1, $2, $3, $4, 'RECEIVED')
                """,
                job_id, document_id, tenant_id, filename
            )

        try:
            # Stage: PARSING
            await self._update_stage(job_id, "PARSING")
            loader = PyPDFLoader(file_path)
            raw_docs = loader.load()

            # Stage: CHUNKING
            await self._update_stage(job_id, "CHUNKING")
            chunks = self.splitter.split_documents(raw_docs)
            await self._set_chunk_count(job_id, len(chunks))

            # Stage: EMBEDDING
            await self._update_stage(job_id, "EMBEDDING")
            texts = [chunk.page_content for chunk in chunks]
            
            # Batch embedding with progress tracking
            embeddings = await self._embed_with_progress(job_id, texts)

            # Stage: INDEXING
            await self._update_stage(job_id, "INDEXING")
            await self._index_chunks(
                conn_pool=self.db,
                job_id=job_id,
                document_id=document_id,
                tenant_id=tenant_id,
                chunks=chunks,
                embeddings=embeddings,
            )

            # Mark READY
            await self._mark_ready(job_id, len(chunks))

            return {
                "job_id": job_id,
                "document_id": document_id,
                "status": "READY",
                "chunk_count": len(chunks),
            }

        except Exception as exc:
            # Capture which stage failed
            async with self.db.acquire() as conn:
                current_stage = await conn.fetchval(
                    "SELECT stage FROM document_ingestion_jobs WHERE job_id = $1",
                    job_id
                )
                await conn.execute(
                    """
                    UPDATE document_ingestion_jobs
                    SET stage = 'FAILED',
                        error_stage = $2,
                        error_message = $3,
                        updated_at = NOW()
                    WHERE job_id = $1
                    """,
                    job_id, current_stage, str(exc)
                )
            raise RuntimeError(
                f"Ingestion failed at stage {current_stage}: {exc}"
            ) from exc

    async def _embed_with_progress(
        self, job_id: str, texts: list[str], batch_size: int = 50
    ) -> list[list[float]]:
        """Embed in batches, updating progress counter after each batch."""
        all_embeddings = []
        embedded_so_far = 0

        for i in range(0, len(texts), batch_size):
            batch = texts[i : i + batch_size]
            batch_embeddings = await self.embedder.aembed_documents(batch)
            all_embeddings.extend(batch_embeddings)
            embedded_so_far += len(batch)

            async with self.db.acquire() as conn:
                await conn.execute(
                    """
                    UPDATE document_ingestion_jobs
                    SET embedded_chunks = $2, updated_at = NOW()
                    WHERE job_id = $1
                    """,
                    job_id, embedded_so_far
                )

        return all_embeddings

    async def _index_chunks(
        self,
        conn_pool: asyncpg.Pool,
        job_id: str,
        document_id: str,
        tenant_id: str,
        chunks,
        embeddings: list[list[float]],
    ):
        """Write chunks and vectors to pgvector, updating indexed_chunks."""
        async with conn_pool.acquire() as conn:
            await register_vector(conn)
            
            for i, (chunk, embedding) in enumerate(zip(chunks, embeddings)):
                await conn.execute(
                    """
                    INSERT INTO document_chunks
                        (chunk_id, document_id, tenant_id, content, embedding, metadata)
                    VALUES ($1, $2, $3, $4, $5, $6)
                    ON CONFLICT (chunk_id) DO UPDATE
                        SET content = EXCLUDED.content,
                            embedding = EXCLUDED.embedding
                    """,
                    str(uuid.uuid4()),
                    document_id,
                    tenant_id,
                    chunk.page_content,
                    embedding,
                    dict(chunk.metadata),
                )

                if (i + 1) % 25 == 0:
                    await conn.execute(
                        """
                        UPDATE document_ingestion_jobs
                        SET indexed_chunks = $2, updated_at = NOW()
                        WHERE job_id = $1
                        """,
                        job_id, i + 1
                    )

    async def _update_stage(self, job_id: str, stage: str):
        async with self.db.acquire() as conn:
            await conn.execute(
                """
                UPDATE document_ingestion_jobs
                SET stage = $2, updated_at = NOW()
                WHERE job_id = $1
                """,
                job_id, stage
            )

    async def _set_chunk_count(self, job_id: str, count: int):
        async with self.db.acquire() as conn:
            await conn.execute(
                "UPDATE document_ingestion_jobs SET chunk_count = $2 WHERE job_id = $1",
                job_id, count
            )

    async def _mark_ready(self, job_id: str, chunk_count: int):
        async with self.db.acquire() as conn:
            await conn.execute(
                """
                UPDATE document_ingestion_jobs
                SET stage = 'READY',
                    indexed_chunks = $2,
                    ready_at = NOW(),
                    updated_at = NOW()
                WHERE job_id = $1
                """,
                job_id, chunk_count
            )

This pipeline does something important: it doesn't return 200 OK until the document is genuinely queryable. Your API endpoint can directly await pipeline.ingest_document(...) and the caller knows the document is ready when they get a response. No polling required.

The tradeoff is obvious — your HTTP connection stays open for 15-30 seconds on a large document. That's acceptable for internal tools or CLI scripts, but not for a user-facing web UI.

---

Async Ingestion with a Job Ticket Pattern

For production web applications, you want to accept the document quickly and process it in the background. The key is giving the caller a job ticket they can use to query readiness.

Here's the FastAPI endpoint layer:

from fastapi import FastAPI, UploadFile, HTTPException, BackgroundTasks
import aiofiles
import tempfile

app = FastAPI()

@app.post("/documents/upload")
async def upload_document(
    file: UploadFile,
    tenant_id: str,
    background_tasks: BackgroundTasks,
    pipeline: IngestionPipeline = Depends(get_pipeline),
):
    # Accept immediately, return job ticket
    document_id = str(uuid.uuid4())
    job_id = str(uuid.uuid4())
    
    # Save file to temp storage
    suffix = Path(file.filename).suffix
    with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
        content = await file.read()
        tmp.write(content)
        tmp_path = tmp.name

    # Create job record synchronously (before background task starts)
    async with app.state.db.acquire() as conn:
        await conn.execute(
            """
            INSERT INTO document_ingestion_jobs
                (job_id, document_id, tenant_id, filename, stage)
            VALUES ($1, $2, $3, $4, 'RECEIVED')
            """,
            job_id, document_id, tenant_id, file.filename
        )

    # Queue background processing
    background_tasks.add_task(
        pipeline.ingest_document,
        file_path=tmp_path,
        tenant_id=tenant_id,
        document_id=document_id,
        job_id=job_id,  # Pass existing job_id to avoid creating a new one
    )

    return {
        "document_id": document_id,
        "job_id": job_id,
        "status": "RECEIVED",
        "poll_url": f"/documents/jobs/{job_id}/status",
    }

@app.get("/documents/jobs/{job_id}/status")
async def get_job_status(job_id: str):
    async with app.state.db.acquire() as conn:
        row = await conn.fetchrow(
            """
            SELECT
                job_id,
                document_id,
                stage,
                chunk_count,
                embedded_chunks,
                indexed_chunks,
                error_stage,
                error_message,
                created_at,
                ready_at
            FROM document_ingestion_jobs
            WHERE job_id = $1
            """,
            job_id
        )

    if not row:
        raise HTTPException(status_code=404, detail="Job not found")

    result = dict(row)
    
    # Compute progress percentage when available
    if result["chunk_count"] and result["indexed_chunks"] is not None:
        result["progress_pct"] = round(
            result["indexed_chunks"] / result["chunk_count"] * 100, 1
        )

    return result

Client-Side Polling

Here's what the client-side polling loop looks like in JavaScript, with exponential backoff:

async function waitForDocumentReady(jobId, maxWaitMs = 120_000) {
  const startTime = Date.now();
  let delay = 500; // start at 500ms

  while (Date.now() - startTime < maxWaitMs) {
    const response = await fetch(`/documents/jobs/${jobId}/status`);
    const job = await response.json();

    if (job.stage === "READY") {
      return { success: true, documentId: job.document_id };
    }

    if (job.stage === "FAILED") {
      throw new Error(
        `Ingestion failed at ${job.error_stage}: ${job.error_message}`
      );
    }

    // Show progress to user
    if (job.progress_pct !== undefined) {
      updateProgressBar(job.progress_pct);
      updateStatusText(`Processing: ${job.stage} (${job.progress_pct}%)`);
    }

    // Exponential backoff, capped at 5s
    await sleep(delay);
    delay = Math.min(delay * 1.5, 5000);
  }

  throw new Error("Document processing timed out");
}

This gives users a responsive experience without hammering your status endpoint. The progress percentage, computed from your indexed_chunks / chunk_count counters, turns "waiting" into something that feels like it's actively working.

---

Real-Time Readiness with Server-Sent Events

Polling is functional but not elegant. For a better user experience, use Server-Sent Events (SSE) to push status updates as they happen:

from fastapi.responses import StreamingResponse
import asyncio
import json

@app.get("/documents/jobs/{job_id}/stream")
async def stream_job_status(job_id: str):
    async def event_generator():
        previous_stage = None
        
        while True:
            async with app.state.db.acquire() as conn:
                row = await conn.fetchrow(
                    """
                    SELECT stage, chunk_count, embedded_chunks, 
                           indexed_chunks, error_stage, error_message
                    FROM document_ingestion_jobs WHERE job_id = $1
                    """,
                    job_id
                )

            if not row:
                yield f"data: {json.dumps({'error': 'Job not found'})}\n\n"
                return

            stage = row["stage"]
            
            # Only emit when something changed
            if stage != previous_stage:
                payload = {
                    "stage": stage,
                    "chunk_count": row["chunk_count"],
                    "embedded_chunks": row["embedded_chunks"],
                    "indexed_chunks": row["indexed_chunks"],
                }

                if stage == "FAILED":
                    payload["error"] = {
                        "stage": row["error_stage"],
                        "message": row["error_message"],
                    }

                yield f"data: {json.dumps(payload)}\n\n"
                previous_stage = stage

            if stage in ("READY", "FAILED"):
                return  # Close the stream

            await asyncio.sleep(0.5)

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )

The X-Accel-Buffering: no header is critical if you're running behind Nginx — without it, Nginx will buffer the SSE stream and your client won't receive events until the buffer fills.

---

Guarding Your Query Endpoint Against Stale Results

Fixing ingestion is only half the problem. You also need to ensure your query endpoint doesn't surface partial results from documents that haven't finished indexing.

If you're using pgvector for similarity search, you can join against your job state table to exclude in-flight documents:

-- Only return chunks from READY documents
SELECT
    dc.chunk_id,
    dc.content,
    dc.metadata,
    1 - (dc.embedding <=> $1::vector) AS similarity
FROM document_chunks dc
INNER JOIN document_ingestion_jobs dij
    ON dij.document_id = dc.document_id
    AND dij.tenant_id = $2
WHERE dij.stage = 'READY'
    AND dc.tenant_id = $2
ORDER BY dc.embedding <=> $1::vector
LIMIT $3;

For managed vector stores like Pinecone or Weaviate, you can't do a SQL join, but you can use metadata filtering to achieve the same effect. Add an index_status metadata field to every chunk at index time:

# When indexing each chunk to Pinecone:
index.upsert(vectors=[{
    "id": chunk_id,
    "values": embedding,
    "metadata": {
        "document_id": document_id,
        "tenant_id": tenant_id,
        "content": chunk_text,
        "index_status": "READY",  # Only set after all chunks are indexed
        "indexed_at": datetime.now(timezone.utc).isoformat(),
    }
}])

Then in your query:

results = index.query(
    vector=query_embedding,
    top_k=10,
    filter={
        "tenant_id": {"$eq": tenant_id},
        "index_status": {"$eq": "READY"},
    }
)

---

Handling Partial Failures and Resumable Ingestion

Your pipeline will fail mid-stream. Embedding API timeouts, vector store rate limits, transient network errors — all of these can leave a document partially indexed. Without a recovery strategy, you end up with orphaned chunk records and a job stuck in EMBEDDING.

The key technique is idempotent chunk writes combined with resumable pipeline logic:

async def resume_failed_job(self, job_id: str):
    """
    Resume an ingestion job from where it failed.
    Safe to call multiple times on the same job.
    """
    async with self.db.acquire() as conn:
        job = await conn.fetchrow(
            """
            SELECT job_id, document_id, tenant_id, filename,
                   stage, error_stage, chunk_count, indexed_chunks
            FROM document_ingestion_jobs
            WHERE job_id = $1 AND stage = 'FAILED'
            """,
            job_id
        )

    if not job:
        raise ValueError(f"No failed job found with id {job_id}")

    # Reset to the failed stage so we retry from there
    failed_at = job["error_stage"]
    
    async with self.db.acquire() as conn:
        await conn.execute(
            """
            UPDATE document_ingestion_jobs
            SET stage = $2, error_stage = NULL, error_message = NULL, updated_at = NOW()
            WHERE job_id = $1
            """,
            job_id, failed_at
        )

    # Re-run from the failed stage
    # The _index_chunks method uses ON CONFLICT DO UPDATE, so re-indexing
    # already-indexed chunks is safe (idempotent)
    return await self.ingest_document(
        file_path=self._get_stored_file_path(job["document_id"]),
        tenant_id=job["tenant_id"],
        document_id=job["document_id"],
        job_id=job_id,
    )

The idempotency here comes from using ON CONFLICT DO UPDATE in the INSERT statement for chunks. Re-indexing a chunk that's already in the vector store just overwrites it with an identical value — no duplicates, no errors.

---

Testing Consistency Guarantees Under Concurrency

You need to actually verify your consistency guarantees hold under realistic conditions. Here's a test harness that simulates concurrent uploads and immediate queries:

import asyncio
import time
import httpx
import pytest

async def upload_and_immediately_query(
    client: httpx.AsyncClient,
    file_path: str,
    query: str,
    tenant_id: str,
) -> dict:
    """
    Upload a document, wait for READY, then query.
    Assert the queried content reflects the uploaded document.
    """
    upload_start = time.monotonic()
    
    with open(file_path, "rb") as f:
        upload_resp = await client.post(
            "/documents/upload",
            files={"file": f},
            data={"tenant_id": tenant_id},
        )
    upload_resp.raise_for_status()
    job = upload_resp.json()
    job_id = job["job_id"]
    document_id = job["document_id"]

    # Poll until READY (with timeout)
    ready = False
    for attempt in range(60):  # 60 * 2s = 2 minute max
        await asyncio.sleep(2)
        status_resp = await client.get(f"/documents/jobs/{job_id}/status")
        status = status_resp.json()
        
        if status["stage"] == "READY":
            ready = True
            ingestion_time = time.monotonic() - upload_start
            break
        elif status["stage"] == "FAILED":
            pytest.fail(f"Ingestion failed: {status['error_message']}")

    assert ready, f"Document not ready within timeout. Last stage: {status['stage']}"

    # Query immediately after READY
    query_resp = await client.post(
        "/query",
        json={"query": query, "tenant_id": tenant_id},
    )
    results = query_resp.json()

    # Assert the uploaded document appears in results
    retrieved_doc_ids = {r["document_id"] for r in results["chunks"]}
    assert document_id in retrieved_doc_ids, (
        f"Document {document_id} not found in query results immediately after READY. "
        f"Found: {retrieved_doc_ids}"
    )

    return {
        "ingestion_time_s": ingestion_time,
        "result_count": len(results["chunks"]),
    }

@pytest.mark.asyncio
async def test_concurrent_read_after_write_consistency():
    """
    Upload 5 documents concurrently and verify each is immediately
    queryable after its individual READY signal.
    """
    test_docs = [
        ("tests/fixtures/security_patch_v1.pdf", "What is the CVE fixed in patch v1?"),
        ("tests/fixtures/api_reference_v3.pdf", "How do I authenticate with API v3?"),
        ("tests/fixtures/runbook_deploy.pdf", "What are the deployment steps?"),
        ("tests/fixtures/changelog_q4.pdf", "What changed in Q4?"),
        ("tests/fixtures/architecture_overview.pdf", "What databases does the system use?"),
    ]

    async with httpx.AsyncClient(base_url="http://localhost:8000") as client:
        tasks = [
            upload_and_immediately_query(client, path, query, tenant_id=f"test-{i}")
            for i, (path, query) in enumerate(test_docs)
        ]
        results = await asyncio.gather(*tasks)

    for i, result in enumerate(results):
        print(f"Doc {i}: ready in {result['ingestion_time_s']:.1f}s, "
              f"{result['result_count']} results")
        assert result["result_count"] > 0

Run this test suite in your CI pipeline. If it fails intermittently, that's your consistency guarantee breaking down under load — typically because your embedding batch job is slower than your polling interval, or because your vector store's write path has eventual consistency behavior of its own.

---

Integrating with Agents That Ingest Their Own Documents

If you're building agentic RAG systems where agents can upload and immediately query documents as part of a workflow, the consistency problem becomes especially sharp. An agent that downloads a report, uploads it to the knowledge base, and then immediately queries it will fail without explicit readiness coordination.

Here's a tool function you'd give an agent:

from langchain.tools import tool

@tool
async def ingest_and_confirm_ready(
    file_path: str,
    query_hint: str,
    tenant_id: str,
) -> str:
    """
    Ingest a document into the knowledge base and wait until it is
    fully indexed and queryable. Returns confirmation with chunk count.
    Use this before attempting to retrieve from a newly added document.
    
    Args:
        file_path: Local path to the document file
        query_hint: What you plan to search for (for logging only)
        tenant_id: The tenant namespace for this document
    """
    pipeline = get_pipeline()  # dependency injection
    
    try:
        result = await pipeline.ingest_document(
            file_path=file_path,
            tenant_id=tenant_id,
        )
        return (
            f"Document successfully indexed. "
            f"document_id={result['document_id']}, "
            f"chunk_count={result['chunk_count']}. "
            f"You may now retrieve from it using tenant_id={tenant_id}."
        )
    except RuntimeError as e:
        return f"Ingestion failed: {e}. Do not attempt retrieval from this document."

The tool's docstring explicitly tells the LLM what this tool does and when to use it. Crucially, it blocks until READY — so by the time the agent sees the confirmation message, the document is actually queryable. Without this guarantee, an agent might see "document accepted" (202 status) and immediately query, finding nothing.

This kind of tool design connects naturally to tool-augmented agent architectures where agents need to manage their own context and information sources.

---

Hands-On Exercise

Build a document freshness monitor that catches consistency violations in an existing RAG system. This is a diagnostic tool — you'll use it to measure the actual read-after-write gap in a real deployment.

Setup: You have an existing RAG system with a PostgreSQL backend and a document upload endpoint. You suspect ingestion is happening asynchronously but you're not sure how long the gap actually is.

Step 1: Add the ingestion job state table from the SQL earlier in this article. If you have an existing pipeline, modify your ingestion code to write state transitions to this table.

Step 2: Write a probe function that:

1. Uploads a small test document (a 1-paragraph text file with a unique UUID as content)
2. Records the timestamp of the successful upload response
3. Polls the query endpoint every 500ms, searching for the UUID
4. Records the timestamp when the UUID first appears in results
5. Computes the gap

import asyncio
import uuid
import time
import httpx

async def measure_read_after_write_gap(base_url: str, tenant_id: str) -> float:
    """
    Returns the read-after-write gap in seconds.
    """
    unique_token = str(uuid.uuid4()).replace("-", "")
    content = f"CONSISTENCY_PROBE_{unique_token}"
    
    # Create a probe document
    with open("/tmp/probe.txt", "w") as f:
        f.write(content)

    async with httpx.AsyncClient(base_url=base_url, timeout=120) as client:
        # Upload
        with open("/tmp/probe.txt", "rb") as f:
            resp = await client.post(
                "/documents/upload",
                files={"file": ("probe.txt", f, "text/plain")},
                data={"tenant_id": tenant_id},
            )
        resp.raise_for_status()
        upload_time = time.monotonic()

        # Poll query endpoint until probe document appears
        for attempt in range(240):  # 2-minute max
            await asyncio.sleep(0.5)
            query_resp = await client.post(
                "/query",
                json={"query": unique_token, "tenant_id": tenant_id},
            )
            results = query_resp.json()
            
            if any(unique_token in r.get("content", "") 
                   for r in results.get("chunks", [])):
                gap = time.monotonic() - upload_time
                return gap

    raise TimeoutError("Probe document never appeared in query results")

# Run several probes and report statistics
async def main():
    import statistics
    gaps = []
    for i in range(10):
        gap = await measure_read_after_write_gap(
            "http://localhost:8000", f"probe-tenant-{i}"
        )
        gaps.append(gap)
        print(f"Probe {i+1}: {gap:.2f}s")

    print(f"\nMean gap: {statistics.mean(gaps):.2f}s")
    print(f"Max gap: {max(gaps):.2f}s")
    print(f"P95 gap: {sorted(gaps)[int(len(gaps) * 0.95)]:.2f}s")

asyncio.run(main())

Step 3: Run this probe before and after implementing the state machine and synchronous readiness signaling from this article. Document the reduction in maximum gap. If your system has an SLA around document availability (e.g., "documents must be searchable within 30 seconds of upload"), this probe becomes a standing integration test.

---

Common Mistakes & Troubleshooting

Mistake: Creating the job record inside the background task You return 202 Accepted with a job_id, but the background task hasn't started yet. Client polls /status and gets 404. Create the record synchronously before queueing the background work, as shown in the FastAPI endpoint above.

Mistake: Marking READY after the last chunk is sent rather than confirmed Your embedding API call is async. You fire off 50 chunks, get a response object back, mark READY — but the actual vector store write is still in flight. Wait for explicit write confirmation from your vector store client before transitioning to READY.

Mistake: Not handling vector store eventual consistency Pinecone's serverless tier and some Weaviate configurations don't guarantee immediate read-after-write at the infrastructure layer. Add a verification step: after marking READY, do a targeted fetch of the first chunk by ID and confirm it's readable before returning to the caller.

# Verification step before marking READY
async def _verify_indexing(self, document_id: str, sample_chunk_id: str):
    """Confirm vector store can actually read back what we just wrote."""
    for attempt in range(10):
        result = index.fetch(ids=[sample_chunk_id])
        if result.vectors.get(sample_chunk_id):
            return True
        await asyncio.sleep(0.5)
    raise RuntimeError("Vector store read-back verification failed")

Mistake: Forgetting to clean up temporary files You write uploaded files to /tmp for processing. Background task runs, ingestion fails at the embedding stage. You retry, but the temp file was cleaned up between attempts. Store files in durable storage with predictable paths before queueing background work.

Mistake: Partial chunk visibility during batch indexing If you're indexing 200 chunks and you don't use index_status: PENDING → READY atomicity, chunks 1-50 become queryable while 51-200 are still being written. Depending on your use case, this may surface dangling or misleading partial results. Decide explicitly whether you want atomic document-level availability or incremental chunk-level availability.

Troubleshooting: Job stuck in EMBEDDING Check your embedding API rate limits. OpenAI's text-embedding-3-small has a default TPM limit that you can easily exceed with large concurrent uploads. Implement exponential backoff on 429 responses in your _embed_with_progress method and ensure your batch size respects the per-request token limits. You can find guidance on context window considerations in the LLM context windows article.

---

Summary & Next Steps

Read-after-write consistency in RAG is a solvable problem — but it requires treating ingestion as a first-class pipeline with explicit state, not a fire-and-forget background job. The core pattern is:

1. Model ingestion as a state machine with a persistent job record tracking every stage transition
2. Choose your consistency mode — synchronous for simplicity, async + polling for scalability, SSE for best UX
3. Guard your query endpoint to filter out documents that haven't reached READY
4. Write idempotent chunk operations so partial failures are safely resumable
5. Measure your actual gap with probe documents before assuming consistency

This architecture becomes especially important as your RAG system grows. When you're ingesting hundreds of documents per hour, the inconsistency windows compound. The state machine approach also gives you visibility into bottlenecks — if P95 ingestion time is 45 seconds and embedding consistently takes 40 of those seconds, you know exactly where to invest (larger batches, a faster embedding model, or a local embedding service).

From here, there are several natural extensions to explore. You can layer metadata filtering on top of your readiness filter to build more sophisticated access controls. You can extend the state machine to support document versioning — tracking not just READY/FAILED but which version of a document is active in the index. And once your ingestion pipeline is solid, the natural next step is evaluating retrieval quality — because consistent availability only matters if what you're retrieving is actually useful.