feat(prisma): parallel fetch — ThreadPoolExecutor fan-out, serialized fallback tier (refs #659)

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

@@ -0,0 +1,371 @@
+# 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.

src/prisma/fetch.py

@@ -27,7 +27,9 @@ import json as _json
 import logging
 import re
 import shutil
+import threading
 import time
+from concurrent.futures import ThreadPoolExecutor, as_completed
 from dataclasses import dataclass
 from pathlib import Path
 from typing import TYPE_CHECKING
@@ -50,6 +52,10 @@ _FALLBACK_MIRRORS = (
     "https://sci-hub.box",
 )
 
+# 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()
+
 
 @dataclass
 class PendingItem:
@@ -403,6 +409,7 @@ def fetch_one(
             return "pmc", path
 
     if client_proxied is not None:
+        with _fallback_lock:
             url = fetch_fallback(client_proxied, doi)
             if url:
                 path = download(client_proxied, url, filename)
@@ -455,6 +462,7 @@ def run(
     fetch_proxy: str | None,
     limit: int | None = None,
     progress: bool = True,
+    workers: int = 8,
 ) -> dict[str, int]:
     """Fetch PDFs for every non-excluded item in the project.
 
@@ -484,16 +492,27 @@ def run(
             proxy=fetch_proxy,
         )
 
+    done = 0
     try:
-        for i, item in enumerate(pending, 1):
+        with ThreadPoolExecutor(max_workers=max(1, workers)) as pool:
-            try:
+            futures = {
-                hit = fetch_one(
+                pool.submit(
+                    fetch_one,
                     item,
                     email=email,
                     scratch=scratch_dir,
                     client=client,
                     client_proxied=client_proxied,
-                )
+                ): item
+                for item in pending
+            }
+            # All DB work stays on this thread — the sqlite connection
+            # must never be shared with the pool.
+            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
@@ -512,9 +531,9 @@ def run(
                         log.warning("attach failed for item %s: %s", item.zot_id, e)
                         stats["errors"] += 1
 
-            if progress and i % 25 == 0:
+                if progress and done % 25 == 0:
                     print(
-                    f"  fetched {i}/{len(pending)}  "
+                        f"  fetched {done}/{len(pending)}  "
                         f"(up={stats['unpaywall']} pmc={stats['pmc']} "
                         f"fb={stats['fallback']} miss={stats['missed']})"
                     )

tests/prisma/test_fetch_deep.py

@@ -2,7 +2,13 @@
 
 from __future__ import annotations
 
-from prisma.fetch import attach_pdf, pending_queue, run
+import threading
+import time
+
+import httpx
+
+from prisma import fetch as fetch_mod
+from prisma.fetch import PendingItem, attach_pdf, fetch_one, pending_queue, run
 from zot.db import TYPE_MAP, Db
 from zot.schema import create_db
 
@@ -33,6 +39,142 @@ class TestPendingQueue:
             assert result == []
 
 
+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
+
+
+def _seed_project(path, n):
+    """Seed *n* screen:include items in project:test into an existing DB."""
+    ids = []
+    with Db(path) as db:
+        for i in range(n):
+            item_id = db.create_item(TYPE_MAP["journalArticle"])
+            db.set_fields(item_id, {"title": f"Article {i}", "DOI": f"10.1/{i}"})
+            db.tag_item(item_id, "project:test")
+            db.tag_item(item_id, "screen:include")
+            ids.append(item_id)
+        db.commit()
+    return ids
+
+
+class TestRunParallel:
+    def test_all_items_attached_and_counted(self, zotero_db, tmp_path, monkeypatch):
+        _seed_project(zotero_db, 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(zotero_db) 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, zotero_db, tmp_path, monkeypatch):
+        _seed_project(zotero_db, 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(zotero_db) 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, zotero_db, tmp_path, monkeypatch):
+        _seed_project(zotero_db, 3)
+
+        def boom(item, *, email, scratch, client, client_proxied=None):
+            raise RuntimeError("kaput")
+
+        monkeypatch.setattr(fetch_mod, "fetch_one", boom)
+        with Db(zotero_db) 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
+
+
 class TestRun:
     def test_empty_queue(self, tmp_path):
         path = str(tmp_path / "z.sqlite")