Problem 128 · Practical builds

Document Ingestion Pipeline with Dedupe and Tombstones

HardPractical multi-stage build

01Problem

Build the ingestion side of a RAG index. Documents arrive with an id and a version, possibly out of order and repeated. Chunk and embed them, never pay to embed the same content twice, re-embed only what changed on update, and handle deletes so late writes cannot resurrect a deleted document.

Stages — finish each before reading the next

1Ingest and chunk

ingest(doc_id, version, text) splits text into paragraph-bounded chunks of at most max_chars and stores an embedding per chunk via an injected embed function.

2Content-hash dedupe

Skip documents whose full-text hash is unchanged. Key chunks by SHA-256 so identical chunks across documents are embedded once and reference-counted.

3Incremental re-chunking

On update, embed only chunk hashes that are new and evict chunks whose reference count drops to zero. Ignore writes whose version is not newer than the stored one.

4Tombstones and deletes

delete(doc_id, version, now) evicts the document's chunks and records a tombstone that rejects older writes, even for a delete that arrives before the create. A newer version resurrects the document; gc_tombstones(now, grace) drops old tombstones.

02Key insight

Make chunk boundaries content-defined (paragraphs) so an edit only shifts local chunks, then make the chunk hash the identity: embeddings become a refcounted content-addressed store. Increment new refs before decrementing old ones so a chunk shared by the old and new version is never evicted and re-embedded. Versions plus tombstones turn an unordered event stream into last-writer-wins.

03Solution

solution.pypython87 lines

import hashlib
from dataclasses import dataclass
from typing import Callable


def sha(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def chunk(text: str, max_chars: int) -> list[str]:
    """Content-defined boundaries (paragraphs), so an edit only changes local chunks."""
    out: list[str] = []
    for para in (p.strip() for p in text.split("\n\n")):
        for i in range(0, len(para), max_chars):
            out.append(para[i:i + max_chars])
    return out


@dataclass(frozen=True)
class Outcome:
    status: str     # new | updated | unchanged | stale | deleted
    embedded: int   # chunks newly embedded (cost)
    evicted: int    # chunks whose last reference went away


class Pipeline:
    def __init__(self, embed: Callable[[str], list[float]], max_chars: int = 200) -> None:
        self.embed, self.max_chars = embed, max_chars
        self.docs: dict[str, tuple[int, str, list[str]]] = {}  # id -> (version, doc hash, chunk hashes)
        self.tombstones: dict[str, tuple[int, float]] = {}      # id -> (version, deleted_at)
        self.vectors: dict[str, list[float]] = {}               # chunk hash -> embedding
        self.texts: dict[str, str] = {}
        self.refs: dict[str, int] = {}

    def _is_stale(self, doc_id: str, version: int) -> bool:
        cur = self.docs.get(doc_id)
        tomb = self.tombstones.get(doc_id)
        return (cur is not None and version <= cur[0]) or (tomb is not None and version <= tomb[0])

    def _incref(self, hashes: list[str], texts: dict[str, str]) -> int:
        embedded = 0
        for h in hashes:
            if self.refs.get(h, 0) == 0:
                self.vectors[h], self.texts[h] = self.embed(texts[h]), texts[h]
                embedded += 1
            self.refs[h] = self.refs.get(h, 0) + 1
        return embedded

    def _decref(self, hashes: list[str]) -> int:
        evicted = 0
        for h in hashes:
            self.refs[h] -= 1
            if self.refs[h] == 0:
                del self.refs[h], self.vectors[h], self.texts[h]
                evicted += 1
        return evicted

    def ingest(self, doc_id: str, version: int, text: str) -> Outcome:
        if self._is_stale(doc_id, version):
            return Outcome("stale", 0, 0)
        cur = self.docs.get(doc_id)
        doc_hash = sha(text)
        if cur is not None and cur[1] == doc_hash:  # whole-document dedupe
            self.docs[doc_id] = (version, doc_hash, cur[2])
            return Outcome("unchanged", 0, 0)
        pieces = chunk(text, self.max_chars)
        texts = {sha(p): p for p in pieces}
        hashes = [sha(p) for p in pieces]
        embedded = self._incref(hashes, texts)        # incref first: shared chunks never hit 0
        evicted = self._decref(cur[2]) if cur else 0
        self.docs[doc_id] = (version, doc_hash, hashes)
        self.tombstones.pop(doc_id, None)             # newer version resurrects
        return Outcome("updated" if cur else "new", embedded, evicted)

    def delete(self, doc_id: str, version: int, now: float) -> Outcome:
        if self._is_stale(doc_id, version):
            return Outcome("stale", 0, 0)
        cur = self.docs.pop(doc_id, None)
        evicted = self._decref(cur[2]) if cur else 0
        self.tombstones[doc_id] = (version, now)      # also blocks late-arriving older writes
        return Outcome("deleted", 0, evicted)

    def gc_tombstones(self, now: float, grace: float) -> int:
        old = [d for d, (_, at) in self.tombstones.items() if now - at >= grace]
        for d in old:
            del self.tombstones[d]
        return len(old)

Try it first.

TimeO(L) per document of length L, plus one embed call per new unique chunk

SpaceO(unique chunks + docs + tombstones)

tests.py · 19 passing

tests.pypython29 lines

calls: list[str] = []
def fake_embed(t: str) -> list[float]:
    calls.append(t)
    return [float(len(t))]
p = Pipeline(fake_embed, max_chars=10)
assert chunk("abc\n\n\n\ndefghijklmnop", 10) == ["abc", "defghijklm", "nop"]
# stage 1: ingest + chunk
o = p.ingest("d1", 1, "alpha\n\nbeta\n\ngamma")
assert o == Outcome("new", 3, 0) and len(p.vectors) == 3
# stage 2: content-hash dedupe (whole doc and cross-doc chunks)
assert p.ingest("d1", 2, "alpha\n\nbeta\n\ngamma") == Outcome("unchanged", 0, 0)
assert p.ingest("d2", 1, "beta\n\ndelta") == Outcome("new", 1, 0)
assert p.refs[sha("beta")] == 2 and len(calls) == 4
# stage 3: incremental re-chunking only embeds what changed
assert p.ingest("d1", 3, "alpha\n\nBETA\n\ngamma") == Outcome("updated", 1, 0)  # "beta" still used by d2
assert p.ingest("d1", 4, "alpha\n\nBETA") == Outcome("updated", 0, 1)          # gamma evicted
assert p.ingest("d1", 3, "late write") == Outcome("stale", 0, 0)
assert calls == ["alpha", "beta", "gamma", "delta", "BETA"]
# stage 4: tombstones
assert p.delete("d2", 2, now=100.0) == Outcome("deleted", 0, 2)   # beta + delta gone
assert sha("beta") not in p.vectors and sha("alpha") in p.vectors
assert p.ingest("d2", 1, "beta\n\ndelta").status == "stale"        # out-of-order replay blocked
assert p.delete("d3", 5, now=100.0).status == "deleted"            # delete before create
assert p.ingest("d3", 4, "early").status == "stale"
assert p.gc_tombstones(now=150.0, grace=100.0) == 0
assert p.gc_tombstones(now=200.0, grace=100.0) == 2
assert p.ingest("d2", 3, "delta").status == "new"                  # newer version resurrects
assert sorted(p.texts.values()) == ["BETA", "alpha", "delta"]
assert all(p.refs[h] >= 1 for h in p.vectors)

04Staff-level follow-up

What breaks at scale? Tombstone GC opens a replay window, so set grace longer than the maximum queue lag; fixed-size windows inside long paragraphs still cascade, so use rolling-hash (FastCDC-style) boundaries; and the refcount store must be updated transactionally with the vector index, typically via an outbox so a crash cannot leak or double-free vectors.