Keeping a RAG Knowledge Base Fresh: Incremental Ingestion, Permissions and Citations

A RAG knowledge base is usually right on the day it is loaded. It goes wrong later, quietly: an edited policy still answers from its old chunk, a deleted document stays retrievable, two ingestion jobs overlap and duplicate the corpus, a finance runbook lands in a support agent's context, a citation points at a chunk that no longer exists, or a new embedding model degrades search without a single error.

None of that is the retriever's fault. It comes from ingestion and the data model. This is the reference implementation I use for that side: Postgres 16 with pgvector, psycopg 3, and tests that run against the real database. If you want the retrieval fundamentals first, start with RAG from first principles in Postgres and pgvector.

It is a minimal version built and tested for this article, not a client system. The embedder is a deterministic hashing stand-in behind LangChain's Embeddings interface; in production the only change is the embedder, for example OpenAIEmbeddings(model="text-embedding-3-small"), which I did not execute here.

The schema: sources, versions, documents, chunks

I model four things: the source (a Confluence space, a Drive folder, a Git repo), each sync run of it, its documents, and their chunks. Embeddings live in a separate table per model.

-- ... sources, and source_versions (source_id, version, status: running/complete/failed)
CREATE TABLE documents (
  id                bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  source_id         bigint NOT NULL REFERENCES sources(id),
  external_id       text   NOT NULL,                -- id in the source system
  title             text   NOT NULL,
  body              text   NOT NULL,
  fingerprint       text   NOT NULL,                -- hash of body + ACL + chunker config
  allowed_groups    text[] NOT NULL,
  last_seen_version int    NOT NULL,
  deleted_at        timestamptz,                    -- tombstone
  UNIQUE (source_id, external_id)
);

CREATE TABLE chunks (
  id             bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  document_id    bigint NOT NULL REFERENCES documents(id) ON DELETE CASCADE,
  chunk_hash     text   NOT NULL,                   -- sha256 of the chunk text
  ordinal        int    NOT NULL,
  section_path   text   NOT NULL,                   -- "Refunds > Enterprise"
  char_start     int    NOT NULL,                   -- offsets into documents.body
  char_end       int    NOT NULL,
  text           text   NOT NULL,
  allowed_groups text[] NOT NULL,                   -- denormalised from the document
  seen_version   int    NOT NULL,
  UNIQUE (document_id, chunk_hash)
);
CREATE INDEX chunks_allowed_groups ON chunks USING gin (allowed_groups);

CREATE TABLE embedding_models (
  name       text PRIMARY KEY,
  dim        int  NOT NULL,
  table_name text NOT NULL UNIQUE,
  status     text NOT NULL CHECK (status IN ('candidate', 'active', 'retired'))
);
-- At most one model serves reads at any moment.
CREATE UNIQUE INDEX one_active_model ON embedding_models ((true)) WHERE status = 'active';

The choices that carry weight:

• UNIQUE (source_id, external_id) makes the source system's id the identity of a document. Re-running a sync can only update, never duplicate.
• UNIQUE (document_id, chunk_hash) makes a chunk's identity its content, not its position. That is what lets unchanged text keep its row and its vector.
• documents.body is kept because citations resolve against the document, not chunks.
• allowed_groups is copied onto every chunk, so the permission check is a column on the row being retrieved.
• The partial unique index makes "two models serving reads" unrepresentable.

A sync is a numbered, locked, restartable run

Every sync of a source gets a new version. Documents and chunks record the version that last saw them. That single integer is what makes deletes safe and re-runs cheap.

Two syncs of the same source must never interleave. If they do, one run's "not seen in this version" looks like a deletion to the other. I use a Postgres advisory lock keyed on the source name:

@contextmanager
def source_lock(conn: psycopg.Connection, source: str):
    """Session-level advisory lock: held across the many transactions of one
    sync, released on unlock or when the connection dies."""
    key = f"kb-sync:{source}"
    got = conn.execute("SELECT pg_try_advisory_lock(hashtextextended(%s, 0))",
                       (key,)).fetchone()[0]
    if not got:
        raise SourceBusy(source)
    try:
        yield
    finally:
        conn.execute("SELECT pg_advisory_unlock(hashtextextended(%s, 0))", (key,))

def sync_source(conn: psycopg.Connection, source: str, docs: Iterable[SourceDoc],
                embedders: Mapping[str, Embeddings], *, max_chars: int = 800,
                max_delete_ratio: float = 0.3, allow_mass_delete: bool = False) -> dict:
    with source_lock(conn, source):
        # ... upsert the sources row, insert source_versions (version = max + 1)
        stats: Counter = Counter()
        try:
            for doc in docs:  # may be a lazy crawl that fails half-way
                stats[upsert_document(conn, source_id, version, doc, embedders,
                                      max_chars, stats)] += 1
            stats["tombstoned"] = tombstone_missing(
                conn, source_id, version, max_delete_ratio, allow_mass_delete)
        except BaseException:
            _finish(conn, source_id, version, "failed", stats)
            raise
        _finish(conn, source_id, version, "complete", stats)
        return dict(stats, version=version)

Why these particular choices:

• `pg_try_advisory_lock`, not the blocking form. A scheduler should skip a run while the previous one is still going, not queue waiters. A test starts four threads syncing the same source and checks the document count stays exact.
• Session-level, not transaction-level. One sync spans many short transactions, so an xact lock would be released at the first commit.
• No lock row with a TTL. If the worker is killed, the server ends its session and frees the lock; a test closes a lock-holding connection and syncs again. A TTL is always too short or too long.
• Caveats: session locks do not work through PgBouncer in transaction-pooling mode, so the worker needs a direct connection. A hashtextextended collision only makes two sources serialise, which is harmless.

Re-running is a no-op: a test syncs twice and asserts the second run returns {"unchanged": 3, "tombstoned": 0, "version": 2} with every chunk id intact. For the queue and retry layer around a job like this, see idempotency, retries and queues with FastAPI and Redis.

Only re-embed what changed

Change detection has two levels. The document fingerprint hashes body, ACL and chunker configuration; if it matches, only last_seen_version moves. If not, the document is re-chunked and each chunk's SHA-256 decides whether it needs a new vector.

    fp = fingerprint(doc, max_chars)
    row = conn.execute("SELECT id, fingerprint, deleted_at FROM documents "
                       "WHERE source_id = %s AND external_id = %s",
                       (source_id, doc.external_id)).fetchone()
    if row and row[1] == fp and row[2] is None:
        conn.execute("UPDATE documents SET last_seen_version = %s WHERE id = %s",
                     (version, row[0]))
        return "unchanged"
    # ... chunk_document(), keeping the first occurrence of each chunk_hash

# Embed only hashes this document does not already have, per live model,
    # and do it before opening the transaction (no network calls inside it).
    new_vectors: dict[str, dict[str, np.ndarray]] = {}
    for model, table in write_models(conn):
        have = {h for (h,) in conn.execute(sql.SQL(
            "SELECT c.chunk_hash FROM chunks c JOIN {t} e ON e.chunk_id = c.id "
            "WHERE c.document_id = %s").format(t=sql.Identifier(table)),
            (row[0] if row else None,))}
        todo = [c for c in chunks if c.chunk_hash not in have]
        vecs = embedders[model].embed_documents([c.text for c in todo]) if todo else []
        new_vectors[table] = {c.chunk_hash: np.asarray(v) for c, v in zip(todo, vecs)}
        stats[f"embedded:{model}"] += len(todo)

with conn.transaction(), conn.cursor() as cur:
        # ... INSERT INTO documents ... ON CONFLICT (source_id, external_id) DO UPDATE
        cur.executemany("""
            INSERT INTO chunks (document_id, chunk_hash, ordinal, section_path, char_start,
                                char_end, text, allowed_groups, seen_version)
            VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
            ON CONFLICT (document_id, chunk_hash) DO UPDATE SET
              ordinal = EXCLUDED.ordinal, section_path = EXCLUDED.section_path,
              char_start = EXCLUDED.char_start, char_end = EXCLUDED.char_end,
              allowed_groups = EXCLUDED.allowed_groups, seen_version = EXCLUDED.seen_version
            """, [(doc_id, c.chunk_hash, i, c.section_path, c.char_start, c.char_end,
                   c.text, list(doc.allowed_groups), version) for i, c in enumerate(chunks)])
        # ... look up chunk ids, INSERT new vectors ON CONFLICT (chunk_id) DO NOTHING
        # Chunks not produced by this version are gone from the source text.
        cur.execute("DELETE FROM chunks WHERE document_id = %s AND seen_version < %s",
                    (doc_id, version))

What each part prevents:

• Re-embedding an unchanged corpus. Changing "14 days" to "30 days" in one paragraph re-embeds exactly one chunk; the test also checks untouched chunks keep their ids and no orphaned vectors remain.
• Readers seeing a half-updated document. Document, chunks, new vectors and stale-chunk removal commit together. Embedding calls happen before BEGIN, so a slow provider never holds a transaction open.
• Stale permissions. An ACL change alters the fingerprint but no chunk hash, so allowed_groups is rewritten on every chunk and nothing is embedded.
• A chunker change that does nothing. Without the chunker config in the fingerprint, unchanged documents would keep their old chunks forever.
• Duplicate paragraphs. Repeated boilerplate yields the same (document_id, chunk_hash) twice. A multi-row INSERT ... ON CONFLICT DO UPDATE then fails with "ON CONFLICT DO UPDATE command cannot affect row a second time"; with executemany the second copy silently overwrites the first one's offsets. I dedupe first.

Hashing only pays off if chunk boundaries depend on content. My chunker packs whole paragraphs and never crosses a heading. On a synthetic 30-section document, inserting one sentence into section 3 changed 1 of 30 structure-aware chunks, but 20 of 21 fixed 400-character windows (50 overlap), because every later boundary shifted.

Deletes are tombstones, written last

A document that leaves the source must stop being retrievable. The dangerous version is "delete whatever this run did not see": a run that saw nothing because the API timed out would wipe the knowledge base.

def tombstone_missing(conn: psycopg.Connection, source_id: int, version: int,
                      max_ratio: float, allow_mass_delete: bool) -> int:
    """Runs only after the full listing succeeded. Documents not seen in this
    version keep their row (id, title, deleted_at) but lose body and chunks."""
    with conn.transaction():
        live, missing = conn.execute(
            "SELECT count(*), count(*) FILTER (WHERE last_seen_version < %s) "
            "FROM documents WHERE source_id = %s AND deleted_at IS NULL",
            (version, source_id)).fetchone()
        if live and missing / live > max_ratio and not allow_mass_delete:
            raise MassDeletionRefused(f"{missing}/{live} documents vanished in one sync")
        ids = [r[0] for r in conn.execute(
            "UPDATE documents SET deleted_at = now(), body = '', fingerprint = '' "
            "WHERE source_id = %s AND deleted_at IS NULL AND last_seen_version < %s "
            "RETURNING id", (source_id, version))]
        conn.execute("DELETE FROM chunks WHERE document_id = ANY(%s)", (ids,))
        return len(ids)

Three guards, each covered by a test:

• Only after a complete listing. If the lazy iterator raises on page two, the version is marked failed and this function never runs. The test checks all documents survive a crawl that dies halfway.
• A mass-deletion guard. An empty listing, usually expired credentials, raises MassDeletionRefused. A human can re-run with allow_mass_delete=True.
• Tombstone the document, hard-delete the chunks. Chunks and, through ON DELETE CASCADE, their vectors go, so deleted text cannot be retrieved. The document row stays with its body cleared, so an old citation can say "removed" rather than 404. If it comes back, the empty fingerprint forces a full re-ingest.

Permissions live in Postgres, not in the prompt

A WHERE allowed_groups && ... clause in application code works until a second query path forgets it. I use Row-Level Security, so every query through the application role is filtered, including the vector search.

ALTER TABLE documents ENABLE ROW LEVEL SECURITY;
ALTER TABLE chunks    ENABLE ROW LEVEL SECURITY;

CREATE POLICY reader_documents ON documents FOR SELECT TO rag_ingestion_reader
  USING (allowed_groups && string_to_array(current_setting('app.user_groups', true), ','));
CREATE POLICY reader_chunks ON chunks FOR SELECT TO rag_ingestion_reader
  USING (allowed_groups && string_to_array(current_setting('app.user_groups', true), ','));
CREATE POLICY ingestor_documents ON documents FOR ALL TO rag_ingestion_ingestor
  USING (true) WITH CHECK (true);
CREATE POLICY ingestor_chunks ON chunks FOR ALL TO rag_ingestion_ingestor
  USING (true) WITH CHECK (true);

Each per-model embedding table gets its own policy, USING (EXISTS (SELECT 1 FROM chunks c WHERE c.id = chunk_id)), so a raw SELECT on the vectors is filtered too.

The details that decide whether this is actually secure:

• A non-superuser role without `BYPASSRLS`. Superusers and BYPASSRLS roles skip every policy silently, as do table owners unless you FORCE ROW LEVEL SECURITY. A test asserts rolsuper = false and rolbypassrls = false for both application roles; another shows the superuser seeing every document while set to ops.
• Transaction-local groups. I call SELECT set_config('app.user_groups', %s, true); the true makes it SET LOCAL. SET LOCAL itself cannot take a bind parameter (psycopg 3 binds server-side, and Postgres rejects $1 there as a syntax error), which tempts people into string formatting. Group names are also validated against [a-z0-9_-], so a comma cannot smuggle in a second group.
• Fail closed. With nothing set, string_to_array yields no groups and the overlap is false. A pooled connection reused by the next request sees nothing until that request sets its own groups.

This is the test I care about most:

PAYROLL_QUERY = "Payroll corrections over one month's salary need CFO sign-off before release."

def test_vector_query_never_returns_another_groups_chunks(reader, embedders, loaded):
    # Control: finance asks the exact payroll sentence and gets it first.
    finance = search(reader, ["finance"], PAYROLL_QUERY, embedders, k=3)
    assert finance[0].document_id == loaded["payroll"]
    # Ops asks the identical question: payroll chunks are not in the candidate set.
    ops = search(reader, ["ops"], PAYROLL_QUERY, embedders, k=50)
    assert ops, "ops still gets its own chunks"
    assert {h.document_id for h in ops} == {loaded["oncall"]}

def test_groups_do_not_leak_to_the_next_transaction(reader, loaded):
    with reader.transaction():
        set_groups(reader, ["finance"])
        assert reader.execute("SELECT count(*) FROM chunks").fetchone()[0] > 0
    # Same pooled connection, next request, nobody set groups: fail closed.
    assert reader.execute("SELECT count(*) FROM chunks").fetchone() == (0,)

The control assertion matters: without it, "ops got no payroll chunks" might only mean the query was bad. Sibling tests query the embedding table directly as ops, ordered by distance to a real payroll vector, and get zero payroll rows; and they check that no groups means no rows and that SET LOCAL with a bind parameter fails.

RLS trusts whatever sets app.user_groups, so groups must come from the verified identity in the API layer, never from the request body or the model. It does not help if an attacker can run arbitrary SQL as the reader role. The same principle, checking permissions in the tool server rather than in the prompt, is in the support agent with tool permissions and human handoff.

There is a recall cost, which I measured. In pgvector 0.6 the HNSW scan returns at most hnsw.ef_search candidates (default 40) and RLS filters them afterwards. On this machine, with 2,000 synthetic chunks of which 200 were visible to ops, a k=10 query returned 1 row at the default and 10 with SET LOCAL hnsw.ef_search = 400. With only 40 of 2,000 chunks visible, the planner used the GIN index on allowed_groups and sorted exactly, with no shortfall. It is a recall problem, not a leak. Fixes: a higher ef_search for sparse groups, partitions or partial indexes per large tenant, or pgvector 0.8's iterative index scans. Measure it with retrieval evals as described in RAG vs long context, with retrieval evals.

Citations that point at the document, not the chunk

A citation stored as a chunk id breaks at the first re-chunk. My anchor is the document id, section path, character offsets into the body, and the quoted text:

@dataclass(frozen=True)
class Anchor:
    document_id: int
    section_path: str
    char_start: int
    char_end: int
    quote: str  # the exact cited text, stored with the answer
    # ... __str__ renders "doc:<id>#<section path>@<start>-<end>"

def resolve(conn: psycopg.Connection, groups: Sequence[str], a: Anchor) -> Resolution:
    with conn.transaction():
        set_groups(conn, groups)  # following a citation is a read: RLS applies
        row = conn.execute("SELECT body, deleted_at FROM documents WHERE id = %s",
                           (a.document_id,)).fetchone()
    if row is None:
        return Resolution("unavailable")  # missing, or not yours: same answer
    body, deleted_at = row
    if deleted_at is not None:
        return Resolution("deleted")
    if body[a.char_start:a.char_end] == a.quote:
        return Resolution("exact", a.char_start, a.char_end)
    hits = [m.start() for m in re.finditer(re.escape(a.quote), body)]
    if not hits:
        return Resolution("stale")  # the cited text no longer exists: re-answer
    in_section = [h for h in hits for path, s, e in sections(body)
                  if path == a.section_path and s <= h < e]
    best = min(in_section or hits, key=lambda h: abs(h - a.char_start))
    return Resolution("moved", best, best + len(a.quote))

Each status maps to a tested scenario:

• exact: ingest with 60-character chunks, cite one sentence, re-ingest with 800. The chunk row is gone; the anchor still resolves exactly.
• moved: a paragraph inserted above shifts offsets; the quote is found again, preferring its section and the nearest position.
• stale: the cited wording changed, so the answer should be regenerated rather than shown with a citation that no longer supports it.
• deleted and unavailable: the tombstone path and the RLS path. Someone outside the group who opens a citation copied from a colleague's transcript gets unavailable, exactly as for a missing document.

Offsets relative to the document also let an answer cite one sentence rather than a whole chunk.

Changing the embedding model without mixing vector spaces

Vectors from two models do not share a space. If dimensions differ, pgvector raises an error. If they match, it silently ranks by a meaningless distance: nothing fails, quality just drops. (With my hashing stand-in, one sentence embedded by two equal-dimension "models" has cosine 0.00; true by construction for a toy, but the principle holds.) One index holds one model's vectors, and the query is embedded by that same model.

Each model gets its own table (emb_<name>) with its own HNSW index and RLS policy, registered in embedding_models. The migration:

1. Register the model as candidate. From then on every sync writes vectors for active and candidate models (the write_models loop above), so new content opens no gap.
2. Backfill in committed batches: chunks with no row in the new table are embedded and inserted ON CONFLICT DO NOTHING. It is resumable; the test's second pass embeds 0.
3. Evaluate retrieval on the candidate before anyone depends on it.
4. Cut over in one transaction that refuses while any chunk lacks a vector:

def cut_over(conn: psycopg.Connection, name: str) -> None:
    """Make `name` the read model. The old one stays a candidate (still
    dual-written), so rolling back is the same call with the old name."""
    with conn.transaction():
        conn.execute("LOCK TABLE embedding_models IN EXCLUSIVE MODE")
        table = conn.execute("SELECT table_name FROM embedding_models WHERE name = %s "
                             "AND status <> 'retired'", (name,)).fetchone()
        if table is None:
            raise CutOverRefused(f"{name} is not registered or is retired")
        if (gap := missing_count(conn, table[0])) > 0:
            raise CutOverRefused(f"{name} is missing vectors for {gap} chunks")
        conn.execute("UPDATE embedding_models SET status = 'candidate' WHERE status = 'active'")
        conn.execute("UPDATE embedding_models SET status = 'active' WHERE name = %s", (name,))

search() reads the active row once and takes both the table and the query embedder from it, so they cannot disagree. Rollback is cut_over(conn, old_name); later, retire the old model and drop its table. I chose a table per model over a new column on chunks:

For a large corpus, build the HNSW index after the backfill; pgvector's own advice is to load data before indexing.

Checklist

• Documents keyed by (source_id, external_id), chunks by (document_id, chunk_hash).
• Fingerprint covers body, ACL and chunker version plus parameters.
• Chunk boundaries follow headings and paragraphs, not fixed windows.
• Embed before the transaction; document, chunks, vectors and stale-chunk delete commit together.
• pg_try_advisory_lock per source, on a direct connection rather than transaction pooling.
• Tombstones only after a complete listing, behind a mass-deletion guard.
• RLS on documents, chunks and every embedding table; roles tested for no superuser and no BYPASSRLS.
• Groups set with set_config(..., true) from verified identity, failing closed when absent.
• A pytest proving another group's chunks never come back from the vector query, with a control.
• hnsw.ef_search sized for the sparsest group, and filtered recall measured.
• Citations as document id, section path, offsets and quote; never a chunk id.
• One table per embedding model: dual-write, backfill, evaluate, cut over in one transaction.

Tested with Python 3.12.9, PostgreSQL 16.2, pgvector 0.6.2, psycopg 3.3.6, pgvector-python 0.5.0, langchain-core 1.6.5, numpy 2.5.3 and pytest 9.1.1 on 2026-09-27.