# PRD — donto distributed embedding fabric (spare-machine worker)

**2026-06-05 · a complete, paste-ready spec for turning any spare computer into an embedding worker for the donto substrate. Built for millions→billions of embeddings.**

> **You are Claude on a spare computer.** Your job: stand up a **donto embedding worker** that pulls un-embedded items from the donto coordinator, computes `bge-small-en-v1.5` vectors, and submits them back. The server side already exists and is live — you only build the worker. Follow §3 (Quick start) to be embedding within minutes; §4–§8 are the full spec, scaling, and ops.

---

## 1. Why this exists

donto holds tens of millions of statements today and is heading for **billions** of embeddings (predicates, entities, memory chunks, and eventually statement-level vectors). Embedding is **CPU-bound** on the main server (`bge-small` fp32 ≈ 3 chunk/s on a 4-vCPU box). A spare machine — especially one with more cores or a GPU — can multiply throughput. This fabric lets **any number of machines** contribute safely and in parallel.

## 2. Architecture (the beautiful part)

```
   ┌──────────────────────────── donto-db server ─────────────────────────────┐
   │  Postgres (donto-pg)                                                       │
   │   • source tables (donto_statement, memory records, predicates, entities) │
   │   • embedding tables (donto_x_memory_chunk_embedding, donto_predicate_…)   │
   │   • donto_embed_queue   ← the work queue (pending|leased|done)             │
   │                                                                            │
   │  embedding COORDINATOR (donto-embed-coordinator.service, :7930)            │
   │   • PROVIDERS: one per embeddable kind (memory_chunk, predicate, …)        │
   │       pending_ids()  → fill the queue        (enqueuer)                    │
   │       build_texts()  → exact text to embed   (owns text-building)          │
   │       upsert()       → write vectors back    (owns DB writes)              │
   │   • leases via  SELECT … FOR UPDATE SKIP LOCKED  → no double-work          │
   │   • bearer-token auth, dim validation, stale-lease reclaim                 │
   └───────────────▲───────────────────────────────────────────────────────────┘
                   │ HTTPS + Bearer token  (https://memories.apexpots.com/embed/*)
                   │
   ┌───────────────┴──── spare machine(s) — YOU build this ────┐
   │  WORKER (donto_embed_worker.py), N processes or GPU       │
   │   loop:  lease → fastembed(bge-small) → submit            │
   │   dumb by design: the coordinator names the model + text  │
   └──────────────────────────────────────────────────────────┘
```

**Design principles**
- **The worker is dumb.** It never touches Postgres, never builds text, never decides what to embed. It leases `{item_id, text, model}`, embeds, submits `{item_id, vector}`. This keeps it trivial, safe, and correct across every embedding kind.
- **The queue + `SKIP LOCKED`** is the billion-scale primitive: hundreds of concurrent workers can lease disjoint batches with zero coordination and zero double-work; crashed leases auto-reclaim after a TTL.
- **Providers generalize it.** Adding a new embeddable kind (e.g. statement-level vectors) is a server-side provider + a registry row — workers need no changes.
- **Security:** the spare never gets DB credentials. One authenticated HTTPS endpoint; writes go only to additive embedding caches (donto invariant I3 — never destructive).

## 3. Quick start (do this)

**Requirements:** the spare machine needs Python 3.9+ and internet. A GPU is optional (big speedup — see §6).

### 3.1 Environment

```bash
python3 -m venv ~/donto-embed && source ~/donto-embed/bin/activate
pip install --upgrade pip
pip install fastembed            # CPU. For NVIDIA GPU: pip install fastembed-gpu
export DONTO_EMBED_URL="https://memories.apexpots.com/embed"
export DONTO_EMBED_TOKEN="<PASTE THE TOKEN THE OPERATOR GAVE YOU>"
```

> The token is **not** in this public document. The operator will paste it to you. Without it every call returns `401`.

### 3.2 Save the worker

Save the script in §5 as `donto_embed_worker.py`.

### 3.3 Sanity check, then run

```bash
# health (no auth) and stats (auth)
curl -s $DONTO_EMBED_URL/health
curl -s -H "Authorization: Bearer $DONTO_EMBED_TOKEN" $DONTO_EMBED_URL/stats

# ask the operator to enqueue work (or do it yourself if you have the token):
curl -s -H "Authorization: Bearer $DONTO_EMBED_TOKEN" -H 'content-type: application/json' \
     -d '{"target":"memory_chunk","limit":50000}' $DONTO_EMBED_URL/enqueue

# run one worker (then scale: launch one per core — see §6)
python donto_embed_worker.py
```

You should see lines like `leased 128 → embedded → submitted 128 (rate X/s, total N)`. Watch `…/stats` — `queue.done` climbs, `embedded` climbs. That's it.

## 4. Protocol (full spec)

All endpoints under `https://memories.apexpots.com/embed`, JSON bodies, header `Authorization: Bearer <token>`.

| method | path | body | returns |
|---|---|---|---|
| GET | `/health` | — | `{ok, targets[]}` (no auth) |
| GET | `/stats` | — | per-target `{source, embedded, queue:{pending,leased,done}}` |
| POST | `/enqueue` | `{target, limit}` | `{enqueued}` — scan source, fill the queue (admin) |
| POST | `/lease` | `{worker_id, n, target?}` | `{batch:[{target,item_id,text,model,dim}]}` |
| POST | `/submit` | `{worker_id, items:[{target,item_id,vector[]}]}` | `{upserted}` |

**Contract**
- `lease` marks items `leased` (with your `worker_id` + timestamp) and returns the **exact text** to embed and the **model** to use. Leases not submitted within the TTL (~15 min) are reclaimed and re-leased — so a worker crash never loses work.
- `submit` validates each vector's dimension against the target (`bge-small` = 384), upserts into the target's embedding table, and marks the queue rows `done`. Wrong-dimension vectors are dropped (logged), not stored.
- Idempotent: re-submitting an item just overwrites its (identical) vector.

## 5. The worker (save as `donto_embed_worker.py`)

```python
#!/usr/bin/env python3
"""donto embedding worker — lease → embed (bge-small) → submit. Dumb by design."""
import os, time, json, urllib.request, urllib.error

URL   = os.environ["DONTO_EMBED_URL"].rstrip("/")
TOKEN = os.environ["DONTO_EMBED_TOKEN"]
N     = int(os.environ.get("EMBED_N", "128"))          # items per lease
TARGET= os.environ.get("EMBED_TARGET") or None          # None = any target
WORKER= os.environ.get("EMBED_WORKER_ID", f"{os.uname().nodename}:{os.getpid()}")
IDLE  = float(os.environ.get("EMBED_IDLE_SLEEP", "5"))   # sleep when queue empty

def call(path, body):
    req = urllib.request.Request(URL + path, data=json.dumps(body).encode(),
        headers={"content-type": "application/json", "Authorization": f"Bearer {TOKEN}"})
    with urllib.request.urlopen(req, timeout=120) as r:
        return json.load(r)

def main():
    from fastembed import TextEmbedding
    models = {}                      # model_name -> TextEmbedding (loaded once each)
    t0, total = time.time(), 0
    print(f"[worker {WORKER}] start → {URL}", flush=True)
    while True:
        try:
            batch = call("/lease", {"worker_id": WORKER, "n": N, "target": TARGET}).get("batch", [])
        except urllib.error.URLError as e:
            print(f"[worker] lease error: {e}; retrying", flush=True); time.sleep(IDLE); continue
        if not batch:
            time.sleep(IDLE); continue
        # group by model (almost always just bge-small)
        by_model = {}
        for b in batch:
            by_model.setdefault(b["model"], []).append(b)
        items = []
        for model_name, rows in by_model.items():
            if model_name not in models:
                print(f"[worker] loading {model_name} …", flush=True)
                models[model_name] = TextEmbedding(model_name=model_name,
                                                   threads=int(os.environ.get("EMBED_THREADS", "0")) or None)
            vecs = list(models[model_name].embed([r["text"] for r in rows], batch_size=64))
            for r, v in zip(rows, vecs):
                items.append({"target": r["target"], "item_id": r["item_id"],
                              "vector": [float(x) for x in v]})
        try:
            res = call("/submit", {"worker_id": WORKER, "items": items})
        except urllib.error.URLError as e:
            print(f"[worker] submit error: {e}; items re-lease after TTL", flush=True); time.sleep(IDLE); continue
        total += res.get("upserted", 0)
        rate = total / max(time.time() - t0, 1e-6)
        print(f"[worker] leased {len(batch)} → submitted {res.get('upserted',0)} "
              f"(rate {rate:.1f}/s, total {total})", flush=True)

if __name__ == "__main__":
    main()
```

## 6. Scaling — millions to billions

- **One process per core.** `bge-small` CPU inference is single-stream; to use all cores, run **N copies** (the queue + `SKIP LOCKED` keeps them from colliding):
  ```bash
  for i in $(seq $(nproc)); do EMBED_WORKER_ID="$(hostname):$i" python donto_embed_worker.py & done
  ```
- **GPU = the real unlock.** `pip install fastembed-gpu` (needs CUDA). One mid-range GPU does **hundreds–thousands of chunk/s** vs ~3/s per CPU core — this is how billions become tractable. Run a single GPU worker with a large `EMBED_N` (e.g. 512–1024).
- **Many machines.** Just run the worker on each; they all lease from the same queue. Linear scaling, no coordination.
- **Throughput math.** ~25k chunks ≈ 2h on one slow CPU core; the same on an 8-core box ≈ 15–20 min; on a GPU ≈ 1–2 min. For a billion at, say, 2,000/s aggregate (a few GPUs) ≈ ~6 days of wall-clock — embarrassingly parallel, so add machines to divide it.
- **Batch size.** Larger `EMBED_N` amortizes HTTP round-trips; 128–512 is a good range (GPU: higher). Text is capped server-side (~300 tokens for chunks), so batches are cheap to move.

## 7. Observability &amp; ops

- `GET /stats` is the dashboard: `source` (total embeddable), `embedded` (done), and live `queue` depth. Poll it to watch progress and compute rate.
- **Verify by the real metric** — watch `embedded` climb over wall-clock, not a worker's self-reported rate. (A worker that prints "rate 200/s" while `embedded` is flat is stuck — check the token / network.)
- **Stale leases self-heal** (TTL ~15 min) — kill a worker any time; its in-flight items return to `pending`.
- **Re-embedding** (e.g. after a chunking change) = the operator re-enqueues that target; workers pick it up.

## 8. Adding new embedding targets (server-side, for reference)

The coordinator is provider-driven. A new kind (e.g. `statement` — billion-scale) is added on the server as a small provider implementing `pending_ids` / `build_texts` / `upsert` over its source + embedding table + model, registered in `PROVIDERS`. Workers need **zero** changes — they already embed whatever `lease` hands them. v1 ships `memory_chunk`; `predicate`, `entity`, and `statement` follow the same pattern.

## 9. Security

- The spare machine never receives Postgres credentials — only an HTTPS endpoint + a bearer token.
- The token gates all reads/writes; rotate it by updating `/etc/donto/embed-coordinator.env` on the server and restarting the service.
- Writes go **only** to additive embedding caches (donto I3 — never destructive to `donto_statement`). A misbehaving worker can at worst submit a bad vector for a leased id (dim-validated, overwsritten on re-embed) — it cannot harm source data.
- Workers do receive item **text** (to embed it), so treat the token as sensitive; share it only with machines you trust.

---

*Server components (operator reference): `donto-embed-coordinator.service` (:7930), `donto_embed_queue` table, `embed_coordinator.py` providers, Caddy path `memories.apexpots.com/embed/* → :7930`. Model is pinned to `BAAI/bge-small-en-v1.5` (384-dim) to match the existing fabric — do not change it without re-embedding everything.*