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

• Step 1: Write the failing test

Append to tests/prisma/test_fetch_deep.py:

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

• 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).

• Step 3: Write minimal implementation

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

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

    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

• Step 4: Run test to verify it passes

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

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

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.

• 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):

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.

• 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).

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

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

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

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

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

• Step 1: Add the option

In the fetch command signature after limit:

    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().

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

• Step 3: Commit

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

• 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).

• Step 2: Relaunch in the background

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

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

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