Enrichment Queue Throughput Proposal

Status: PROPOSAL ONLY — no changes until approved. Date: 2026-06-08. Author: M.P. Goal: Make the direct-to-gpumon throughput we are seeing during the drain the permanent steady state, surfaced on gpumon's enrichment-depth panel, by re-activating the existing broker path (option C) — now that the false-skip bug that motivated the drain is fixed at every layer.

TL;DR

1. The system you want already exists and is deployed. enrich-scheduler (producer) → enrich.<task> RabbitMQ queues → enrich-runner (5 replicas, consumer) → enrich.dlx DLQ. The drain (backfill()) was always an explicit stopgap "before the U3 runner / U4 scheduler land." They landed. The drain only runs because the broker path was throttled/starved after the false-skip incident.
2. A producer that dumps 10,000 envelopes never waits for the LLM. enqueue() is pure Postgres + AMQP publish (persistent:true). The runner — not gpumon — pulls. prefetch × replicas is the single saturation knob that decouples generation latency from LLM latency.
3. The "false-skip-prone broker" is fixed at all three layers (task rethrow, llm.ts 120 s timeout, runner error → retry → DLQ with MAX_ATTEMPTS=5 + exponential backoff). Option C is now safe.
4. The drain/scheme workflow is untouched. Re-activating the broker path is a config + redeploy of enrich-scheduler/enrich-runner only. backfill(), regen-driver.ts, scheme-classify.ts are not modified.
5. Verify with a 200 rpm × 5 min envelope load test reusing the fraud-publisher publish/consume harness against a benign __loadtest__ task path. Pass = sustained ack ≥ 200/min, bounded queue depth, zero real-DLQ growth.

1. Why the work isn't on gpumon's depth panel today

The drain calls backfill() → selects() directly, then fires chat() straight at gpumon-ingress:4001. It bypasses RabbitMQ and task_state by design (that was the point — route around the throttled broker). So:

• gpumon sees a stream of synchronous /v1/chat/completions calls with x-gpumon attribution — it load-balances them, but there is no queue depth to display, because the depth lives in the driver's in-process worker pool (N=8), not in a broker gpumon can see.
• The progress board shows the drain's heartbeat rows (extract-case-facts, etc.) because the driver POSTs them — but that is the board, not gpumon's enrichment-depth panel.

The depth panel is fed by RabbitMQ queue metrics (enrich.<task> ready/unacked). To light it up, the work must flow through the broker — i.e. option C.

2. The decoupling you described — it's the existing producer/queue/puller split

"a script can generate 10000 envelopes at once … the throughput of the LLM is not gating and the script is not needing to run slower or wait for responses"

That is exactly enrich-scheduler.enqueue():

enqueue(task, key, trigger):
  if task_state(key) in {queued, running}        → skip (already in flight)
  if task_state(key) in {failed, skipped} && backing off → skip
  markQueued(key); publishJob(ch, {task, target_key:key})   # persistent AMQP, returns immediately

• Producer side (generation): bounded only by Postgres + AMQP publish speed. Dump 10 k envelopes in seconds. Never blocks on an LLM response.
• Queue: enrich.<task>, durable, persistent messages survive broker restart.
• Consumer side (LLM saturation): enrich-runner, 5 replicas, each ch.prefetch(PREFETCH). In-flight LLM requests = PREFETCH × replicas. That — not the producer — is what pulls "just as fast as the LLM pool can sustain."
• gpumon does NOT pull. It is a synchronous LiteLLM-style proxy. The runner pulls from RabbitMQ and makes the synchronous call into gpumon; gpumon's pool router then spreads each call across GPU backends. Two schedulers compose cleanly: RabbitMQ prefetch caps concurrency into gpumon; gpumon spreads that concurrency across the pool.

Saturation knob math. To hold the ≤ 8 LLM-in-flight cap: PREFETCH × replicas ≤ 8. With 5 replicas that forces PREFETCH=1 (5 in-flight) or pin to fewer replicas. To target a higher sustained rate for a dedicated load test, raise the cap deliberately (see §5). Today's stack file says PREFETCH=6 × 5 = 30 — far over the ≤ 8 cap; an earlier container inspection showed PREFETCH=2. This discrepancy must be resolved before re-activation (see §4, step 0).

3. The false-skip fix — why option C is now safe

The drain was built to avoid the broker because the broker path was branding extractable docs as permanently skipped. That root cause is fixed at all three layers:

The runner's makeHandler mapping is the literal verification:

try   → persistTaskRow → "skipped" ? markState(skipped)+return skipped : markState(done)+return done
catch → markState(failed, bumpAttempts) → return {status:"retry"}

consumeTask then republishes with x-attempt+1 until MAX_ATTEMPTS=5, after which it nacks to enrich.<task>.dlq. backoffSeconds(attempt)=min(15m, 30·2^(attempt-1)). A transient LLM error can no longer become a permanent skip — it retries with backoff, then dead-letters for inspection/replay. dlq-replay (currently 0/0) re-injects DLQ messages up to DLQ_MAX_REPLAYS=3.

4. Concrete changes (drain-untouching)

All changes are confined to the two services + their stack env. backfill(), regen-driver.ts, scheme-classify.ts are not edited. Deploy via the no-registry path (docker save | ssh node-eighteen docker load → service update --force), grep a marker string in the running container to confirm.

Step 0 — Measure live config first (currently unconfirmed; stack file vs. earlier inspection disagree). Write a temp .ts and run it inside the runner container (avoid inline bun -e quoting hell):

• live PREFETCH and SWEEP env on the running scheduler/runner;
• queue depths: ch.checkQueue("enrich.extract-case-facts") etc. (ready/unacked);
• consumer counts.

Step 1 — Reconcile the concurrency cap. Decide PREFETCH × replicas for steady state. Recommended steady state: replicas=5, PREFETCH=1 → 5 in-flight (under ≤ 8 cap, leaves headroom for scheme-classify's own in-flight). Update docker/stack-enrich.yml to match reality (kill the PREFETCH=6 value).

Step 2 — Ensure SWEEP=1 on enrich-scheduler (stack says 1; earlier inspection suggested 0). With SWEEP_LIMIT=5000 pushed into SQL (the OOM fix), the backfill sweep is safe to leave on. This is what makes the scheduler continuously refill the queue from the skipped/pending backlog — the always-on behaviour you want.

Step 3 — Rebuild + redeploy the fixed enrich-runner and enrich-scheduler images to node-eighteen, force-update, confirm the false-skip-fix marker is present in the running container.

Step 4 — Point the drain backlog at the broker. Reset the target skipped rows skipped → pending (scoped, in chunks) so the scheduler's sweep enqueues them. This replaces the manual drain going forward — but only after the load test passes; until then the drain keeps running untouched.

5. Load test: sustain 200 rpm for 5 min

Reuse the services/fraud-publisher/smoke.ts + smoke-drain.ts publish/consume pattern. Use a dedicated benign task so production rows are never touched.

Harness (services/fraud-publisher/loadtest.ts, new — does NOT touch drain/scheme):

1. Assert topology for a throwaway task __loadtest__ (its own queue + DLQ).
2. Producer: publish 200 envelopes/min for 5 min = 1,000 envelopes, all up front in the first seconds if desired (proves generation never gates) OR paced — both modes supported via a flag. Envelope marker: task="__loadtest__", target_key = synthetic id.
3. Consumer: a load-test runner whose handler either (a) calls gpumon with a tiny fixed prompt (true end-to-end LLM saturation test) or (b) sleeps a fixed LATENCY_MS (pure broker/throughput test). Run (b) first to prove broker throughput, then (a) to prove the pool sustains it.
4. Measure for the full 5 min: published/sec, acked/sec, queue depth (ready+unacked) every 5 s, DLQ count, p50/p95 handler latency.

Pass criteria:

• sustained ack rate ≥ 200/min averaged over the 5 min, no downward drift;
• queue depth bounded (rises to the in-flight ceiling then plateaus — does not grow monotonically → consumer keeps up);
• zero __loadtest__.dlq growth (no spurious failures);
• gpumon enrichment-depth panel shows the enrich.__loadtest__ depth live (confirms observability goal).
• Production queues (enrich.extract-case-facts etc.) and the live drain/scheme runs show no regression during the test.

Teardown: purge + delete enrich.__loadtest__ and its DLQ; the synthetic marker (task="__loadtest__") guarantees nothing leaks into task_state/case_facts.

Note on 200 rpm vs ≤ 8 cap: 200 requests/min at, say, 1.5 s p50 handler latency needs ≈ 5 concurrent (200/60 × 1.5 ≈ 5). That fits inside the ≤ 8 cap. If real LLM latency is higher, either the rate or the concurrency cap must rise for the test window — surface this explicitly rather than silently exceeding the guard.

6. What stays exactly as-is

• enrich-tasks/src/index.ts backfill(), regen-driver.ts, drain2.ts, scheme-classify.ts — unchanged.
• The running drain (regen-drain2-20260608) and scheme pass (scheme-classify-20260608) — keep running until the load test passes and Step 4 cuts over.
• All LLM traffic stays through gpumon-ingress (pool names only, bearer unchanged). Compute stays on node-eighteen. DB writes stay in the runner container's PG client.

Open items before implementation (require a live probe — proposal only, not done here)

• Confirm live PREFETCH / SWEEP on the deployed scheduler/runner (Step 0).
• Confirm live enrich.<task> queue depths + consumer counts.
• Confirm gpumon's enrichment-depth panel data source is RabbitMQ queue metrics (assumed; verify the panel query).