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

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

# 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>.

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)

#!/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):

  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 & 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.