Lab 11 · CDC → Vector Sync

You will wire a DuckDB source table to a Chroma index through a content-hash CDC loop that detects inserts, updates, and deletes, re-embeds only what changed, and propagates deletions — then prove the pipeline with an out-of-process canary that checks index/source parity, freshness lag, and live retrievability of a newly inserted document. A seeded change set (2 inserts, 3 updates, 1 delete) must sync in a single pass, and breaking any leg of the canary must turn it red. Stale indexes are the silent killer of production RAG: the retrieval code keeps returning 200s while answering from last month's documents, and nothing alerts until a user notices. This lab builds the muscle that prevents that.

Time & tier

~150 minutes hands-on · Advanced tier (maps to module M7).

Setup

Laptop-only, $0, no cloud account. You need the Lab 09 corpus as seed data:

mkdir -p ~/labs/de-lab-11 && cd ~/labs/de-lab-11
python3 -m venv .venv && source .venv/bin/activate
pip install duckdb chromadb sentence-transformers
cp ~/labs/de-lab-09/make_corpus.py .
python make_corpus.py    # writes corpus/docs.jsonl (the seed for the source table)

Create init_source.py — it builds the DuckDB "operational" table your imaginary CMS writes to:

import duckdb, json, pathlib

con = duckdb.connect("plantops.duckdb")
con.execute("""CREATE TABLE IF NOT EXISTS kb_articles (
  id VARCHAR PRIMARY KEY, title VARCHAR, part_code VARCHAR, text VARCHAR,
  updated_at TIMESTAMP DEFAULT current_timestamp)""")
if con.execute("SELECT count(*) FROM kb_articles").fetchone()[0] == 0:
    docs = [json.loads(l) for l in pathlib.Path("corpus/docs.jsonl").read_text().splitlines()]
    con.executemany("INSERT INTO kb_articles (id, title, part_code, text) VALUES (?, ?, ?, ?)",
                    [(d["id"], d["title"], d["part_code"], d["text"]) for d in docs])
print("source rows:", con.execute("SELECT count(*) FROM kb_articles").fetchone()[0])

Run python init_source.py — expect source rows: 120.

Guided steps

1. 1.

   Write the sync engine. Create sync.py. The heart is sync_pass: snapshot the source, compare content hashes against a sync_state table, and derive three sets — inserts (in source, not in state), updates (hash changed), deletes (in state, not in source). Deletes are applied to the index before upserts.

   Copy

   import duckdb, hashlib, json, time, pathlib, sys
   from datetime import datetime
   from sentence_transformers import SentenceTransformer
   import chromadb
   
   def content_hash(title, text):
       return hashlib.sha256(f"{title}\n{text}".encode()).hexdigest()
   
   def sync_pass(con, col, model):
       t0 = time.perf_counter()
       src = {r[0]: (r[1], r[2]) for r in
              con.execute("SELECT id, title, text FROM kb_articles").fetchall()}
       state = {r[0]: r[1] for r in
                con.execute("SELECT id, content_hash FROM sync_state").fetchall()}
       inserts = [i for i in src if i not in state]
       updates = [i for i in src if i in state and state[i] != content_hash(*src[i])]
       deletes = [i for i in state if i not in src]
       if deletes:
           col.delete(ids=deletes)
           con.executemany("DELETE FROM sync_state WHERE id = ?", [(i,) for i in deletes])
       changed = inserts + updates
       if changed:
           texts = [src[i][0] + "\n" + src[i][1] for i in changed]
           emb = model.encode(texts, normalize_embeddings=True, show_progress_bar=False)
           col.upsert(ids=changed, embeddings=emb.tolist(), documents=texts)
           now = datetime.now()
           con.executemany("INSERT OR REPLACE INTO sync_state VALUES (?, ?, ?)",
                           [(i, content_hash(*src[i]), now) for i in changed])
       print(f"sync: +{len(inserts)} ~{len(updates)} -{len(deletes)} "
             f"in {time.perf_counter()-t0:.1f}s; index={col.count()}")
   
   if __name__ == "__main__":
       con = duckdb.connect("plantops.duckdb")
       con.execute("""CREATE TABLE IF NOT EXISTS sync_state (
           id VARCHAR PRIMARY KEY, content_hash VARCHAR, synced_at TIMESTAMP)""")
       model = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")
       col = chromadb.PersistentClient(path="chroma_store").get_or_create_collection(
           "plantops", metadata={"hnsw:space": "cosine"})
       every = int(sys.argv[sys.argv.index("--every") + 1]) if "--every" in sys.argv else 0
       while True:
           sync_pass(con, col, model)
           if not every: break
           time.sleep(every)
   

   Two details deserve a pause. First, timestamps for sync_state come from Python's datetime.now() rather than DuckDB's current_timestamp — DuckDB returns timezone-aware timestamps to Python via pytz, which you may not have installed; generating the timestamp client-side sidesteps the dependency. Second, INSERT OR REPLACE makes the state write idempotent, so a crashed pass simply redoes work on the next run instead of corrupting state.

2. 2.

   Cold-load and prove idempotency. Run python sync.py twice. Pass 1 must report +120 ~0 -0 (cold load, ~1 s of embedding). Pass 2 must report +0 ~0 -0 in well under a second — the no-op pass is the property that makes a tight scheduler loop affordable. If pass 2 re-embeds anything, your hash input differs between passes; find the nondeterminism before continuing.

3. 3.

   Seed a realistic change set. Create seed_changes.py with exactly this mix — two new articles, three revisions, one takedown:

   Copy

   import duckdb
   con = duckdb.connect("plantops.duckdb")
   con.execute("""INSERT INTO kb_articles (id, title, part_code, text) VALUES
     ('kb-500', 'Troubleshooting: laser cutter misalignment (part LSR-1000-A)', 'LSR-1000-A',
      'Applies to laser cutter units, service part LSR-1000-A. Reported symptom: beam drifts off the cut line. Resolution: realign the gantry rails and recalibrate the optics.'),
     ('kb-501', 'Troubleshooting: pallet wrapper film tension (part PWR-2000-B)', 'PWR-2000-B',
      'Applies to pallet wrapper units, service part PWR-2000-B. Reported symptom: film snaps at high tension. Resolution: replace the tension clutch and reset the dancer arm.')""")
   con.execute("""UPDATE kb_articles
     SET text = text || ' REVISION 2: also check the firmware version before replacing hardware.',
         updated_at = current_timestamp
     WHERE id IN ('kb-010', 'kb-011', 'kb-012')""")
   con.execute("DELETE FROM kb_articles WHERE id = 'kb-000'")
   print("seeded: 2 inserts, 3 updates, 1 delete")
   

   Run it, then python sync.py once more. Expected: sync: +2 ~3 -1 … index=121 (120 − 1 + 2).

4. 4.

   Build the canary — as a separate process. Create canary.py. It opens everything fresh, checks three invariants, and exits non-zero on any failure so a scheduler can alert on it:

   Copy

   import duckdb, chromadb, sys
   from sentence_transformers import SentenceTransformer
   
   con = duckdb.connect("plantops.duckdb", read_only=True)
   col = chromadb.PersistentClient(path="chroma_store").get_collection("plantops")
   
   src_n = con.execute("SELECT count(*) FROM kb_articles").fetchone()[0]
   idx_n = col.count()
   lag = con.execute("""SELECT coalesce(max(epoch(a.updated_at) - epoch(s.synced_at)), 0)
     FROM kb_articles a JOIN sync_state s USING (id)""").fetchone()[0]
   
   model = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")
   qv = model.encode(["laser cutter beam drifting off the cutting line"], normalize_embeddings=True)
   top = col.query(query_embeddings=qv.tolist(), n_results=3)["ids"][0]
   
   ok_parity = src_n == idx_n
   ok_fresh = lag <= 900          # 15-minute freshness SLA
   ok_canary = "kb-500" in top
   print(f"parity: source={src_n} index={idx_n} {'OK' if ok_parity else 'FAIL'}")
   print(f"freshness: max unsynced lag {lag:.0f}s (SLA 900s) {'OK' if ok_fresh else 'FAIL'}")
   print(f"canary top-3: {top} {'OK' if ok_canary else 'FAIL'}")
   sys.exit(0 if (ok_parity and ok_fresh and ok_canary) else 1)
   

   Run python canary.py; echo "exit: $?". Expect all three OK, kb-500 at rank 1, exit 0.

5. 5.

   Understand why the canary is out-of-process — this is not cosmetic. Try querying for the laser-cutter document from inside the sync process, immediately after the pass that upserted it alongside a delete. On current Chroma builds you will intermittently find the fresh vector missing from results even though col.get(ids=["kb-500"]) returns it and a fresh process ranks it #1 — the writer's in-memory HNSW view can serve stale reads right after a mutation batch that includes deletes. The lesson generalizes: a monitor that shares state with the pipeline it monitors will lie to you. Production canaries run against the same read path your RAG application uses (separate process, separate connection), never against the writer's handles.

6. 6.

   Run the scheduler loop and measure end-to-end lag. Start python sync.py --every 60 in one terminal. In another, insert a new row into kb_articles (copy the pattern from seed_changes.py with a new id), note the wall-clock time, and poll python canary.py until parity and the new document's retrievability go green. Record the observed lag — it should be under 90 seconds (one polling interval plus embed time), comfortably inside the 15-minute SLA the canary encodes.

7. 7.

   Prove all three failure modes fail loudly. One at a time: (a) delete a source row while the loop is stopped — canary parity FAILs; (b) stop the loop, update a row, wait, and check the freshness leg (or temporarily set the SLA to 1 s to see it trip); (c) col.delete(ids=["kb-500"]) directly — the retrievability leg FAILs. Restore each (run a sync pass) and confirm green. A monitor you have never seen red is untested code.

8. 8.

   Map it to production orchestration. Write production-notes.md (5–10 lines): the while True loop becomes an Airflow DAG on a */15 schedule (or a Dagster sensor on the CDC stream); sync_pass is a task with retries and an SLA; canary.py is a downstream task whose non-zero exit pages on-call; content hashing is replaced by Debezium/warehouse streams at scale, keeping the identical insert/update/delete application logic; and versioned index cutover (build plantops_v2, validate with the canary, flip an alias) replaces in-place sync when you change chunking or embedding models — because re-embedding in place would serve a half-migrated index to users.

Verification

• •
  Cold sync: +120 ~0 -0; second pass: +0 ~0 -0 and fast.
• •
  After seed_changes.py: exactly +2 ~3 -1, index count 121.
• •
  canary.py exits 0 with kb-500 in the top-3 for the laser-cutter query.
• •
  Each of the three sabotage tests in step 7 makes the canary exit 1, and a sync pass restores exit 0.
• •
  epoch(updated_at) - epoch(synced_at) for the revised rows is negative or near zero after sync (synced after updated).

Completion artifact

A de-lab-11/ folder containing init_source.py, sync.py, seed_changes.py, canary.py, and freshness-report.md — the sync-pass log lines (cold, no-op, seeded), the canary output, the measured insert-to-retrievable lag from step 6 demonstrating sub-15-minute freshness, and production-notes.md. Self-grading rubric (pass = 8/10):

• Change detection correct (0–3): cold/no-op/seeded passes report exactly the expected counts, including the delete = 3.
• Canary design (0–3): out-of-process, three independent legs (parity, freshness, retrievability), non-zero exit on failure, and all three failure modes demonstrated red-then-green = 3.
• Measured freshness (0–2): end-to-end lag measured against the scheduler loop and reported against the 15-minute SLA = 2.
• Production mapping (0–1): Airflow/Debezium equivalents and versioned cutover noted = 1.
• Idempotency (0–1): second pass is a no-op; state writes use INSERT OR REPLACE = 1.

Stuck?

• •
  pytz ModuleNotFoundError: you reintroduced current_timestamp fetching through DuckDB into Python — use datetime.now() as in the listing, or pip install pytz.
• •
  Pass 2 keeps re-embedding: your hash covers updated_at or you rebuilt strings differently — hash exactly title + "\n" + text.
• •
  Canary cannot find kb-500 even out-of-process: confirm the sync pass after seeding actually ran against chroma_store (PersistentClient path typo is the usual cause).
• •
  DuckDB "database is locked": DuckDB allows one writer — stop the --every loop before running seed_changes.py, or use separate invocations as in the steps.
• •
  Index count is 120 after seeding: your deletes ran but inserts did not — check that seed_changes.py committed (DuckDB auto-commits per execute; a typo in the INSERT would have raised).

Answer key (don't read before attempting)

1. 1Seeded change set: inserts kb-500, kb-501; updates kb-010, kb-011, kb-012 (REVISION 2 suffix changes the content hash); delete kb-000. Correct single-pass result: +2 ~3 -1, index 121.
2. 2The trap in step 5: querying the writer's own collection handle immediately after a mixed upsert+delete pass can return a stale view that omits the fresh vectors — while a fresh process sees them at rank 1. The canary is out-of-process precisely so it exercises the reader path.
3. 3Failure-mode expectations: source-side delete without sync → parity FAIL (121 vs 120); stopped loop + source update → freshness FAIL once lag exceeds the SLA; index-side tamper delete → retrievability FAIL with parity also off by one. Any leg you could not turn red indicates a canary bug, not pipeline health.

Client-safe variant

The entire pipeline — source database, hashing, embeddings, vector store, canary — runs locally with no external calls, so it can be lifted unchanged into a client VPC where data egress is prohibited. On engagements, the pattern to propose is exactly what you built: CDC from the client's system of record (their Debezium/stream infrastructure if it exists, content hashing if not), embedding via a model hosted inside their boundary, and a canary wired into their existing alerting. The freshness SLA and the red/green evidence from step 7 make a strong client deliverable: you are not just shipping a sync job, you are shipping proof that staleness, deletion leaks, and silent index drift will page someone before a user finds them.