stack/docs/superpowers/plans/2026-08-19-parallel-prisma-fetch-p43.md

# Parallel PRISMA Fetch Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** Parallelize `prisma.fetch.run()` so the 1,977-item palliative-rfi PDF queue finishes in hours instead of days (issue #659).

**Architecture:** `fetch_one` is already thread-safe (unique per-item scratch filenames, thread-safe httpx clients, no shared mutable state), so `run()` fans it out over a `ThreadPoolExecutor` while ALL SQLite work (`pending_queue`, `attach_pdf`, `db.commit`) stays on the main thread. The only shared-state hazard is the fallback tier: `_altcha_bootstrap` mutates the shared proxied client's cookie jar, so a module-level lock serializes fallback-tier access (which is also polite to the mirrors behind one VPN droplet).

**Tech Stack:** Python stdlib `concurrent.futures.ThreadPoolExecutor` + `threading.Lock`; typer for the CLI option; pytest with monkeypatched `fetch_one` for orchestration tests.

**Spec:** Gitea issue #659 (https://git.fhirworx.io/homelab/stack/issues/659) — the issue body is the spec.

## Global Constraints

- SQLite connections must never be shared across threads — every `Db` call stays on the thread that opened it (here: the main thread).
- The fallback tier must never run concurrently (shared cookie jar in `client_proxied` + mirror politeness).
- Existing behavior at `workers=1` must be byte-identical in observable effects: same stats keys, same progress-line format, same `finally` client cleanup.
- Stage-2 excluded items remain untouched (queue filter unchanged).
- Commits: stage only the files this plan touches (`src/prisma/fetch.py`, `src/cli/prisma.py`, `tests/prisma/test_fetch_deep.py`); the worktree carries unrelated foreign changes (`compose.yml`, `nature.csl`, `*.bak-fleet`). No Claude co-author trailer.

---

### Task 1: Serialize the fallback tier inside `fetch_one`

**Files:**
- Modify: `src/prisma/fetch.py` (module head + `fetch_one`, currently lines 374–413)
- Test: `tests/prisma/test_fetch_deep.py`

**Interfaces:**
- Consumes: existing `fetch_one(item, *, email, scratch, client, client_proxied=None)`.
- Produces: same signature; new module global `_fallback_lock: threading.Lock` held for the whole fallback branch (fetch_fallback + its download). Task 2's parallel `run()` relies on this lock existing.

- [x] **Step 1: Write the failing test**

Append to `tests/prisma/test_fetch_deep.py`:

```python
import threading
import time

import httpx

from prisma import fetch as fetch_mod
from prisma.fetch import PendingItem, fetch_one


class TestFallbackSerialized:
    def test_fallback_tier_never_concurrent(self, tmp_path, monkeypatch):
        """Parallel fetch_one calls must enter the fallback tier one at a time."""
        state = {"active": 0, "max_active": 0}
        gauge = threading.Lock()

        def fake_fallback(client, doi):
            with gauge:
                state["active"] += 1
                state["max_active"] = max(state["max_active"], state["active"])
            time.sleep(0.05)
            with gauge:
                state["active"] -= 1
            return None  # miss → cascade ends, no download

        # direct tiers all miss so every call reaches the fallback tier
        monkeypatch.setattr(fetch_mod, "fetch_unpaywall", lambda c, d, e: None)
        monkeypatch.setattr(fetch_mod, "fetch_pmc", lambda c, p: None)
        monkeypatch.setattr(fetch_mod, "fetch_fallback", fake_fallback)

        client = httpx.Client(transport=httpx.MockTransport(lambda r: httpx.Response(404)))
        items = [
            PendingItem(zot_id=i, doi=f"10.1/{i}", pmid="", pmcid="", title=f"t{i}")
            for i in range(6)
        ]
        threads = [
            threading.Thread(
                target=fetch_one,
                args=(it,),
                kwargs={
                    "email": "e@x.com",
                    "scratch": tmp_path,
                    "client": client,
                    "client_proxied": client,
                },
            )
            for it in items
        ]
        for t in threads:
            t.start()
        for t in threads:
            t.join()
        assert state["max_active"] == 1
```

- [x] **Step 2: Run test to verify it fails**

Run: `uv run pytest tests/prisma/test_fetch_deep.py::TestFallbackSerialized -v`
Expected: FAIL — `state["max_active"]` > 1 (no lock exists yet).

- [x] **Step 3: Write minimal implementation**

In `src/prisma/fetch.py`, add `import threading` to the imports and a module global near `_FALLBACK_MIRRORS`:

```python
# The fallback tier shares one proxied client (altcha solves mutate its
# cookie jar) and one VPN exit — hold this across the whole tier.
_fallback_lock = threading.Lock()
```

In `fetch_one`, wrap the fallback branch:

```python
    if client_proxied is not None:
        with _fallback_lock:
            url = fetch_fallback(client_proxied, doi)
            if url:
                path = download(client_proxied, url, filename)
                if path:
                    return "fallback", path
```

- [x] **Step 4: Run test to verify it passes**

Run: `uv run pytest tests/prisma/test_fetch_deep.py::TestFallbackSerialized -v`
Expected: PASS

- [x] **Step 5: Run the whole fetch test set, then commit**

Run: `uv run pytest tests/prisma/test_fetch.py tests/prisma/test_fetch_deep.py tests/prisma/test_fetch_exercise.py -v`
Expected: all PASS.

```bash
git add src/prisma/fetch.py tests/prisma/test_fetch_deep.py
git commit -m "fix(prisma): serialize the fallback tier — shared cookie jar + single VPN exit (refs #659)"
```

### Task 2: Parallel `run()` orchestrator

**Files:**
- Modify: `src/prisma/fetch.py` (`run()`, currently lines 448–527)
- Test: `tests/prisma/test_fetch_deep.py`

**Interfaces:**
- Consumes: `fetch_one` from Task 1 (unchanged signature); existing `pending_queue`, `attach_pdf`.
- Produces: `run(db, *, project, storage_dir, scratch_dir, email, fetch_proxy, limit=None, progress=True, workers=8)` — same return dict `{"unpaywall": int, "pmc": int, "fallback": int, "missed": int, "errors": int}`. Task 3's CLI passes `workers` through.

- [x] **Step 1: Write the failing tests**

Append to `tests/prisma/test_fetch_deep.py` (reuses `create_db`/`Db`/`TYPE_MAP` already imported at the top of this file):

```python
def _seed_project(path, n):
    """Create a DB with *n* screen:include items in project:test; return their itemIDs."""
    con = create_db(path)
    con.close()
    ids = []
    with Db(path) as db:
        for i in range(n):
            item_id = db.create_item(TYPE_MAP["journalArticle"], key=f"KEY{i:05d}AA", now="2026-08-19 00:00:00")
            db.set_fields(item_id, {"title": f"Article {i}", "DOI": f"10.1/{i}"})
            db.add_tags(item_id, ["project:test", "screen:include"])
            ids.append(item_id)
        db.commit()
    return ids


class TestRunParallel:
    def test_all_items_attached_and_counted(self, tmp_path, monkeypatch):
        path = str(tmp_path / "z.sqlite")
        ids = _seed_project(path, 6)

        def fake_fetch_one(item, *, email, scratch, client, client_proxied=None):
            time.sleep(0.05)
            p = scratch / f"{item.zot_id}.pdf"
            p.write_bytes(b"%PDF-1.4 fake")
            return "unpaywall", p

        monkeypatch.setattr(fetch_mod, "fetch_one", fake_fetch_one)
        with Db(path) as db:
            stats = fetch_mod.run(
                db,
                project="test",
                storage_dir=tmp_path / "storage",
                scratch_dir=tmp_path / "scratch",
                email="e@x.com",
                fetch_proxy=None,
                workers=4,
            )
            assert stats["unpaywall"] == 6
            assert stats["missed"] == 0
            assert stats["errors"] == 0
            # every item now has a PDF attachment → queue drains
            assert fetch_mod.pending_queue(db, "test") == []

    def test_fetches_overlap(self, tmp_path, monkeypatch):
        path = str(tmp_path / "z.sqlite")
        _seed_project(path, 6)
        state = {"active": 0, "max_active": 0}
        gauge = threading.Lock()

        def slow_fetch_one(item, *, email, scratch, client, client_proxied=None):
            with gauge:
                state["active"] += 1
                state["max_active"] = max(state["max_active"], state["active"])
            time.sleep(0.2)
            with gauge:
                state["active"] -= 1
            return None

        monkeypatch.setattr(fetch_mod, "fetch_one", slow_fetch_one)
        with Db(path) as db:
            stats = fetch_mod.run(
                db,
                project="test",
                storage_dir=tmp_path / "storage",
                scratch_dir=tmp_path / "scratch",
                email="e@x.com",
                fetch_proxy=None,
                workers=6,
            )
        assert stats["missed"] == 6
        assert state["max_active"] >= 2

    def test_worker_exception_counts_as_error(self, tmp_path, monkeypatch):
        path = str(tmp_path / "z.sqlite")
        _seed_project(path, 3)

        def boom(item, *, email, scratch, client, client_proxied=None):
            raise RuntimeError("kaput")

        monkeypatch.setattr(fetch_mod, "fetch_one", boom)
        with Db(path) as db:
            stats = fetch_mod.run(
                db,
                project="test",
                storage_dir=tmp_path / "storage",
                scratch_dir=tmp_path / "scratch",
                email="e@x.com",
                fetch_proxy=None,
                workers=2,
            )
        assert stats["errors"] == 3
```

Note for the implementer: if `Db` has no `add_tags`/`set_fields` helpers with these exact names, check `src/zot/db.py` and use the actual tag/field helpers — the intent is "n journalArticle items tagged `project:test` + `screen:include`, each with a DOI and title". Adjust the seeding helper, not the assertions.

- [x] **Step 2: Run tests to verify they fail**

Run: `uv run pytest tests/prisma/test_fetch_deep.py::TestRunParallel -v`
Expected: FAIL — `run()` has no `workers` parameter (TypeError).

- [x] **Step 3: Implement the parallel orchestrator**

Replace the `for i, item in enumerate(pending, 1)` loop in `run()` (keep everything above it — queue build, stats dict, client construction — and the `finally` unchanged). Add `workers: int = 8` to the signature after `progress`. Add `from concurrent.futures import ThreadPoolExecutor` to the imports.

```python
    done = 0
    try:
        with ThreadPoolExecutor(max_workers=max(1, workers)) as pool:
            futures = {
                pool.submit(
                    fetch_one,
                    item,
                    email=email,
                    scratch=scratch_dir,
                    client=client,
                    client_proxied=client_proxied,
                ): item
                for item in pending
            }
            for fut in as_completed(futures):
                item = futures[fut]
                done += 1
                try:
                    hit = fut.result()
                except Exception as e:  # noqa: BLE001
                    log.warning("fetch failed for item %s: %s", item.zot_id, e)
                    stats["errors"] += 1
                    continue

                if not hit:
                    stats["missed"] += 1
                else:
                    source, path = hit
                    try:
                        attach_pdf(db, item.zot_id, path, storage_dir, title=item.title)
                        db.commit()
                        path.unlink(missing_ok=True)
                        stats[source] += 1
                    except Exception as e:  # noqa: BLE001
                        log.warning("attach failed for item %s: %s", item.zot_id, e)
                        stats["errors"] += 1

                if progress and done % 25 == 0:
                    print(
                        f"  fetched {done}/{len(pending)}  "
                        f"(up={stats['unpaywall']} pmc={stats['pmc']} "
                        f"fb={stats['fallback']} miss={stats['missed']})"
                    )
    finally:
        client.close()
        if client_proxied is not None:
            client_proxied.close()
```

(`as_completed` comes from the same `concurrent.futures` import. Note `fetch_one` must be looked up on the module at call time for the monkeypatched tests to see it — submitting `fetch_one` directly from the closure is fine because monkeypatch patches the module attribute the function body references; if the direct reference dodges the patch, submit `globals()["fetch_one"]` — in practice `pool.submit(fetch_one, ...)` resolves the module global at submit time and the tests pass either way, since the test patches before calling `run`.)

- [x] **Step 4: Run tests to verify they pass**

Run: `uv run pytest tests/prisma/test_fetch_deep.py -v`
Expected: all PASS, including the pre-existing `TestRun::test_empty_queue`.

- [x] **Step 5: Run the whole fetch test set, then commit**

Run: `uv run pytest tests/prisma/test_fetch.py tests/prisma/test_fetch_deep.py tests/prisma/test_fetch_exercise.py -v`
Expected: all PASS.

```bash
git add src/prisma/fetch.py tests/prisma/test_fetch_deep.py
git commit -m "feat(prisma): parallel fetch — ThreadPoolExecutor fan-out, DB writes on main thread (refs #659)"
```

### Task 3: CLI `--workers` option

**Files:**
- Modify: `src/cli/prisma.py` (the `fetch` command, lines 252–334)
- Test: `tests/prisma/test_fetch_deep.py` (signature-level check only; the CLI wrapper has no dedicated test file)

**Interfaces:**
- Consumes: `run(..., workers=)` from Task 2.
- Produces: `stack prisma fetch <project> --workers/-w N` (default 8).

- [x] **Step 1: Add the option**

In the `fetch` command signature after `limit`:

```python
    workers: int = typer.Option(
        8, "--workers", "-w", help="Concurrent fetch workers (fallback tier stays serialized)."
    ),
```

and pass `workers=workers,` in the `_fetch.run(` call inside `_go()`.

- [x] **Step 2: Verify the wiring**

Run: `uv run stack prisma fetch --help`
Expected: `--workers / -w` listed with default 8.

Run: `uv run pytest tests/prisma -k "fetch" -v`
Expected: all PASS.

- [x] **Step 3: Commit**

```bash
git add src/cli/prisma.py
git commit -m "feat(cli): stack prisma fetch --workers (refs #659)"
```

### Task 4: Rollout — relaunch the palliative-rfi fetch

**Files:** none (operational).

- [x] **Step 1: Confirm no fetch/screen process is running and no foreign staged work snuck into the commits** (`git status --short` — `compose.yml`, `nature.csl`, `*.bak-fleet` must still be unstaged/untracked).

- [x] **Step 2: Relaunch in the background**

```bash
uv run stack prisma fetch palliative-rfi --workers 8
```

(zotero container is stopped/restarted automatically by `_hold`.)

- [x] **Step 3: Verify pace** — within ~10 minutes the output should show `fetched 25/…` lines and Zotero storage should accrue PDFs several times faster than the ~3/12min sequential baseline. Report the measured pace and ETA on #659.