fix(zot): fetch review pass — quota 403 degrades instead of aborting, per-item isolation with rollback and scratch cleanup, ledger notes with an 'error' outcome that retries next run (refs Z1)

- Anna's 403 whose body is not a key/membership error disables Anna's for
  the run (annas_left=0) instead of raising AnnasAuthError.
- run() isolates each item: any exception rolls the transaction back,
  unlinks scratch files, ledgers outcome=error with the exception text,
  and continues; AnnasAuthError still aborts.
- attach_file writes rows before copying so a failed copy rolls back
  cleanly instead of orphaning a storage dir.
- Notes (miss vs error) from resolve/deliver land in every ledger line;
  an item whose lookups/deliveries errored (transport, 5xx, throttle
  page) is 'error' and retried next run rather than a 30-day miss.
- resolve tolerates non-dict file entries and non-numeric filesize.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

src/zot/fetch/attach.py

@@ -57,13 +57,13 @@ def attach_file(
 
     Provenance: tag ``source:<source>`` on the parent and, when an md5
     is known, an ``md5: <md5>`` line in ``extra``. Nothing else on the
-    parent changes.
+    parent changes. The caller commits; on any exception it must roll
+    the transaction back.
     """
     ctype = CONTENT_TYPES.get(path.suffix.lower())
     if ctype is None:
         raise ValueError(f"unsupported attachment type: {path.suffix!r} ({path.name})")
     att_key = generate_key()
-    copier(path, Path(storage_dir) / att_key)
     att_id = db.add_attachment(
         parent_id, key=att_key, content_type=ctype, path=f"storage:{path.name}"
     )
@@ -71,4 +71,7 @@ def attach_file(
     db.tag_item(parent_id, f"source:{source}")
     if md5:
         append_extra(db, parent_id, f"md5: {md5}")
+    # Rows first, file second: a failed copy leaves rows the caller rolls
+    # back; the only orphan case is a failed commit after a good copy.
+    copier(path, Path(storage_dir) / att_key)
     return att_id

src/zot/fetch/deliver.py

@@ -16,13 +16,14 @@ from pathlib import Path
 
 import httpx
 
-from zot.fetch.resolve import LIBGEN, Candidate, Pacer
+from zot.fetch.resolve import LIBGEN, Candidate, Notes, Pacer
 
 ANNAS = "https://annas-archive.gl"
 MIN_BYTES = 10_240
 MAGIC = {"pdf": b"%PDF", "epub": b"PK"}
 _GET_LINK = re.compile(r"get\.php\?md5=([0-9a-f]{32})&(?:amp;)?key=([A-Za-z0-9]+)")
 _ADS_RETRY_SECONDS = 3.0
+_KEY_ERRORS = ("secret key", "not a member", "invalid key", "membership")
 
 
 class AnnasAuthError(RuntimeError):
@@ -50,33 +51,43 @@ def download(
     *,
     ext: str,
     headers: dict[str, str] | None = None,
+    notes: Notes | None = None,
+    what: str = "download",
 ) -> Path | None:
     """Stream *url* to *dest*; None unless it is a real file of type *ext*."""
+    notes = notes if notes is not None else Notes()
     try:
         with client.stream(
             "GET", url, headers=headers, timeout=120, follow_redirects=True
         ) as r:
             if r.status_code != 200:
+                if r.status_code >= 500:
+                    notes.error(f"{what}: http {r.status_code}")
+                else:
+                    notes.miss(f"{what}: http {r.status_code}")
                 return None
             if "text/html" in r.headers.get("content-type", "").lower():
+                notes.miss(f"{what}: answered HTML, not a file")
                 return None
             dest.parent.mkdir(parents=True, exist_ok=True)
             with dest.open("wb") as fh:
                 for chunk in r.iter_bytes(1 << 16):
                     fh.write(chunk)
-    except (httpx.HTTPError, OSError):
+    except (httpx.HTTPError, OSError) as exc:
         dest.unlink(missing_ok=True)
+        notes.error(f"{what}: {type(exc).__name__}")
         return None
     size = dest.stat().st_size
     with dest.open("rb") as fh:
         head = fh.read(4)
     if size < MIN_BYTES or not head.startswith(MAGIC[ext]):
         dest.unlink(missing_ok=True)
+        notes.miss(f"{what}: not a {ext} ({size} bytes)")
         return None
     return dest
 
 
-def _ads_key(client: httpx.Client, md5: str, pacer: Pacer) -> str | None:
+def _ads_key(client: httpx.Client, md5: str, pacer: Pacer, notes: Notes) -> str | None:
     pacer.wait()
     try:
         r = client.get(
@@ -85,12 +96,20 @@ def _ads_key(client: httpx.Client, md5: str, pacer: Pacer) -> str | None:
             headers={"Referer": f"{LIBGEN}/"},
             timeout=30,
         )
-    except httpx.HTTPError:
+    except httpx.HTTPError as exc:
+        notes.error(f"libgen ads: {type(exc).__name__}")
         return None
     if r.status_code != 200:
+        notes.error(f"libgen ads: http {r.status_code}")
         return None
     m = _GET_LINK.search(r.text)
-    return m.group(2) if m and m.group(1) == md5 else None
+    if m is None:
+        notes.error("libgen ads: no get link (throttle page?)")
+        return None
+    if m.group(1) != md5:
+        notes.miss("libgen ads: page links a different md5")
+        return None
+    return m.group(2)
 
 
 def fetch_libgen(
@@ -100,12 +119,14 @@ def fetch_libgen(
     *,
     pacer: Pacer,
     sleep: Callable[[float], None],
+    notes: Notes | None = None,
 ) -> Path | None:
+    notes = notes if notes is not None else Notes()
     md5 = cand.md5 or ""
-    key = _ads_key(client, md5, pacer)
+    key = _ads_key(client, md5, pacer, notes)
     if key is None:
         sleep(_ADS_RETRY_SECONDS)
-        key = _ads_key(client, md5, pacer)
+        key = _ads_key(client, md5, pacer, notes)
     if key is None:
         return None
     return download(
@@ -114,38 +135,65 @@ def fetch_libgen(
         dest,
         ext=cand.ext,
         headers={"Referer": f"{LIBGEN}/ads.php?md5={md5}"},
+        notes=notes,
+        what="libgen get",
     )
 
 
+def _error_text(r: httpx.Response) -> str:
+    try:
+        data = r.json()
+    except ValueError:
+        return r.text[:200]
+    return str(data.get("error") or "")[:200] if isinstance(data, dict) else ""
+
+
 def fetch_annas(
-    client: httpx.Client, cand: Candidate, dest: Path, *, budget: Budget
+    client: httpx.Client,
+    cand: Candidate,
+    dest: Path,
+    *,
+    budget: Budget,
+    notes: Notes | None = None,
 ) -> Path | None:
+    notes = notes if notes is not None else Notes()
     try:
         r = client.get(
             f"{ANNAS}/dyn/api/fast_download.json",
             params={"md5": cand.md5 or "", "key": budget.annas_key},
             timeout=30,
         )
-    except httpx.HTTPError:
+    except httpx.HTTPError as exc:
+        notes.error(f"annas: {type(exc).__name__}")
         return None
     if r.status_code in (401, 403):
-        raise AnnasAuthError(
-            f"Anna's Archive refused the key ({r.status_code}): {r.text[:200]}"
-        )
+        err = _error_text(r)
+        if r.status_code == 401 or any(k in err.lower() for k in _KEY_ERRORS):
+            raise AnnasAuthError(
+                f"Anna's Archive refused the key ({r.status_code}): {err}"
+            )
+        # 403 for anything else (e.g. "No downloads left") → quota exhausted
+        budget.annas_disabled = True
+        budget.annas_left = 0
+        notes.error(f"annas: {err or 'http 403'} — disabled for this run")
+        return None
     if r.status_code not in (200, 204):
         budget.annas_disabled = True
+        notes.error(f"annas: http {r.status_code} — disabled for this run")
         return None
     try:
         data = r.json()
     except ValueError:
+        notes.error("annas: non-JSON answer")
         return None
     info = data.get("account_fast_download_info") or {}
     if isinstance(info.get("downloads_left"), int):
         budget.annas_left = info["downloads_left"]
     url = data.get("download_url")
     if not url:
+        notes.miss(f"annas: {data.get('error') or 'no download_url'}")
         return None
-    return download(client, url, dest, ext=cand.ext)
+    return download(client, url, dest, ext=cand.ext, notes=notes, what="annas get")
 
 
 def fetch(
@@ -157,19 +205,25 @@ def fetch(
     budget: Budget,
     pacer: Pacer,
     sleep: Callable[[float], None] = time.sleep,
+    notes: Notes | None = None,
 ) -> str | None:
     """Deliver *cand* to *dest*; returns the source that worked, else None.
 
     Sets ``budget.quota_hit`` when Anna's would have been tried but the
     reserve (or an earlier API failure) blocked it.
     """
+    notes = notes if notes is not None else Notes()
     if cand.source == "unpaywall":
-        got = download(direct, cand.url or "", dest, ext=cand.ext)
+        got = download(
+            direct, cand.url or "", dest, ext=cand.ext, notes=notes, what="unpaywall"
+        )
         return "unpaywall" if got else None
-    if fetch_libgen(proxied, cand, dest, pacer=pacer, sleep=sleep):
+    if fetch_libgen(proxied, cand, dest, pacer=pacer, sleep=sleep, notes=notes):
         return "libgen"
     if budget.can_use_annas():
-        return "annas" if fetch_annas(proxied, cand, dest, budget=budget) else None
+        got = fetch_annas(proxied, cand, dest, budget=budget, notes=notes)
+        return "annas" if got else None
     if budget.annas_key:
         budget.quota_hit = True
+        notes.miss("annas: skipped (daily reserve or disabled)")
     return None

src/zot/fetch/resolve.py

@@ -28,6 +28,29 @@ _YEAR_RE = re.compile(r"\b(19|20)\d\d\b")
 _EXT_ORDER = {"book": ("epub", "pdf"), "paper": ("pdf",)}
 
 
+class Notes:
+    """Why an item got no file: ``miss`` = a definitive answer (no edition,
+    404, bad file); ``error`` = the service never answered (transport,
+    5xx, throttle page). An item whose notes hold any error is retried
+    next run instead of waiting out the 30-day miss rule."""
+
+    def __init__(self) -> None:
+        self.items: list[tuple[str, str]] = []
+
+    def miss(self, text: str) -> None:
+        self.items.append(("miss", text))
+
+    def error(self, text: str) -> None:
+        self.items.append(("error", text))
+
+    @property
+    def had_error(self) -> bool:
+        return any(kind == "error" for kind, _ in self.items)
+
+    def __str__(self) -> str:
+        return "; ".join(f"{kind}: {text}" for kind, text in self.items)
+
+
 @dataclass(frozen=True)
 class Candidate:
     source: str  # unpaywall | libgen
@@ -63,15 +86,26 @@ class Pacer:
 
 
 def _get_json(
-    client: httpx.Client, url: str, params: dict[str, str], pacer: Pacer
+    client: httpx.Client,
+    url: str,
+    params: dict[str, str],
+    pacer: Pacer,
+    notes: Notes | None = None,
 ) -> dict:
+    notes = notes if notes is not None else Notes()
+    what = f"libgen {params.get('object', '?')}"
     pacer.wait()
     try:
         r = client.get(url, params=params, timeout=30)
         if r.status_code != 200:
+            notes.error(f"{what}: http {r.status_code}")
             return {}
         data = r.json()
-    except (httpx.HTTPError, ValueError):
+    except httpx.HTTPError as exc:
+        notes.error(f"{what}: {type(exc).__name__}")
+        return {}
+    except ValueError:
+        notes.error(f"{what}: non-JSON answer (missing browser UA?)")
         return {}
     return data if isinstance(data, dict) else {}
 
@@ -83,6 +117,7 @@ def libgen_editions(
     isbn: str = "",
     ids: Iterable[str] = (),
     pacer: Pacer,
+    notes: Notes | None = None,
 ) -> dict:
     params = {"object": "e", "addkeys": "*"}
     if doi:
@@ -91,18 +126,36 @@ def libgen_editions(
         params["isbn"] = isbn
     else:
         params["ids"] = ",".join(ids)
-    return _get_json(client, f"{LIBGEN}/json.php", params, pacer)
+    return _get_json(client, f"{LIBGEN}/json.php", params, pacer, notes)
 
 
-def libgen_files(client: httpx.Client, f_ids: list[str], *, pacer: Pacer) -> dict:
+def libgen_files(
+    client: httpx.Client,
+    f_ids: list[str],
+    *,
+    pacer: Pacer,
+    notes: Notes | None = None,
+) -> dict:
     if not f_ids:
         return {}
     params = {"object": "f", "addkeys": "*", "ids": ",".join(f_ids)}
-    return _get_json(client, f"{LIBGEN}/json.php", params, pacer)
+    return _get_json(client, f"{LIBGEN}/json.php", params, pacer, notes)
+
+
+def _int_or_none(value: object) -> int | None:
+    try:
+        return int(value) if value else None
+    except (TypeError, ValueError):
+        return None
 
 
 def _candidates_from_editions(
-    client: httpx.Client, editions: dict, *, kind: str, pacer: Pacer
+    client: httpx.Client,
+    editions: dict,
+    *,
+    kind: str,
+    pacer: Pacer,
+    notes: Notes | None = None,
 ) -> list[Candidate]:
     files: list[tuple[str, str, str]] = []  # (f_id, md5, title)
     for ed in editions.values():
@@ -112,10 +165,14 @@ def _candidates_from_editions(
         raw = ed.get("files")
         entries = raw.values() if isinstance(raw, dict) else (raw or [])
         for f in entries:
+            if not isinstance(f, dict):
+                continue
             md5 = (f.get("md5") or "").lower()
             if md5:
                 files.append((str(f.get("f_id") or ""), md5, title))
-    info = libgen_files(client, [fid for fid, _, _ in files if fid], pacer=pacer)
+    info = libgen_files(
+        client, [fid for fid, _, _ in files if fid], pacer=pacer, notes=notes
+    )
     out: list[Candidate] = []
     for fid, md5, title in files:
         meta = info.get(fid) or {}
@@ -124,22 +181,36 @@ def _candidates_from_editions(
             ext = "pdf"
         if ext not in _EXT_ORDER[kind]:
             continue
-        size = meta.get("filesize")
-        out.append(
-            Candidate("libgen", md5, None, ext, int(size) if size else None, title)
-        )
+        size = _int_or_none(meta.get("filesize"))
+        out.append(Candidate("libgen", md5, None, ext, size, title))
     order = _EXT_ORDER[kind]
     out.sort(key=lambda c: order.index(c.ext))
     return out
 
 
 def libgen_candidates(
-    client: httpx.Client, *, kind: str, doi: str = "", isbn: str = "", pacer: Pacer
+    client: httpx.Client,
+    *,
+    kind: str,
+    doi: str = "",
+    isbn: str = "",
+    pacer: Pacer,
+    notes: Notes | None = None,
 ) -> list[Candidate]:
-    editions = libgen_editions(client, doi=doi, isbn=isbn, pacer=pacer)
+    notes = notes if notes is not None else Notes()
+    editions = libgen_editions(client, doi=doi, isbn=isbn, pacer=pacer, notes=notes)
     if not editions:
+        if not notes.had_error:
+            notes.miss(
+                f"libgen: no edition for {'doi ' + doi if doi else 'isbn ' + isbn}"
+            )
         return []
-    return _candidates_from_editions(client, editions, kind=kind, pacer=pacer)
+    out = _candidates_from_editions(
+        client, editions, kind=kind, pacer=pacer, notes=notes
+    )
+    if not out and not notes.had_error:
+        notes.miss("libgen: edition has no usable file")
+    return out
 
 
 def _norm(s: str) -> str:
@@ -171,8 +242,14 @@ def title_matches(target: str, year: str, entries: list[dict]) -> list[dict]:
 
 
 def title_candidates(
-    client: httpx.Client, *, title: str, year: str, pacer: Pacer
+    client: httpx.Client,
+    *,
+    title: str,
+    year: str,
+    pacer: Pacer,
+    notes: Notes | None = None,
 ) -> list[Candidate]:
+    notes = notes if notes is not None else Notes()
     if not title:
         return []
     pacer.wait()
@@ -185,16 +262,23 @@ def title_candidates(
     }
     try:
         r = client.get(f"{LIBGEN}/index.php", params=params, timeout=30)
-    except httpx.HTTPError:
+    except httpx.HTTPError as exc:
+        notes.error(f"libgen search: {type(exc).__name__}")
         return []
     if r.status_code != 200:
+        notes.error(f"libgen search: http {r.status_code}")
         return []
     entries = [{"id": i, "title": t.strip()} for i, t in _EDITION_RE.findall(r.text)]
     hits = title_matches(title, year, entries)
     if not hits:
+        notes.miss("libgen search: no title match")
         return []
-    editions = libgen_editions(client, ids=[h["id"] for h in hits], pacer=pacer)
-    return _candidates_from_editions(client, editions, kind="book", pacer=pacer)
+    editions = libgen_editions(
+        client, ids=[h["id"] for h in hits], pacer=pacer, notes=notes
+    )
+    return _candidates_from_editions(
+        client, editions, kind="book", pacer=pacer, notes=notes
+    )
 
 
 def candidates(
@@ -205,22 +289,35 @@ def candidates(
     email: str,
     up_pacer: Pacer,
     lg_pacer: Pacer,
+    notes: Notes | None = None,
 ) -> list[Candidate]:
     """Ordered candidates for one item: Unpaywall URLs, then libgen md5s."""
+    notes = notes if notes is not None else Notes()
     out: list[Candidate] = []
     if ref.doi:
         up_pacer.wait()
-        for url in unpaywall_urls(direct, ref.doi, email):
+        urls = unpaywall_urls(direct, ref.doi, email)
+        if not urls:
+            notes.miss("unpaywall: no OA location (or unreachable)")
+        for url in urls:
             out.append(Candidate("unpaywall", None, url, "pdf", None, ref.title))
         out.extend(
-            libgen_candidates(proxied, kind=ref.kind, doi=ref.doi, pacer=lg_pacer)
+            libgen_candidates(
+                proxied, kind=ref.kind, doi=ref.doi, pacer=lg_pacer, notes=notes
+            )
         )
     elif ref.isbn:
         out.extend(
-            libgen_candidates(proxied, kind=ref.kind, isbn=ref.isbn, pacer=lg_pacer)
+            libgen_candidates(
+                proxied, kind=ref.kind, isbn=ref.isbn, pacer=lg_pacer, notes=notes
+            )
         )
     elif ref.kind == "book" and ref.title:
         out.extend(
-            title_candidates(proxied, title=ref.title, year=ref.year, pacer=lg_pacer)
+            title_candidates(
+                proxied, title=ref.title, year=ref.year, pacer=lg_pacer, notes=notes
+            )
         )
+    else:
+        notes.miss("no identifiers (no DOI/ISBN; not a book with a title)")
     return out

src/zot/fetch/run.py

@@ -13,9 +13,9 @@ import httpx
 
 from zot.db import Db
 from zot.fetch.attach import Copier, attach_file, sudo_copier
-from zot.fetch.deliver import Budget, fetch
+from zot.fetch.deliver import AnnasAuthError, Budget, fetch
 from zot.fetch.ledger import Ledger
-from zot.fetch.resolve import BROWSER_UA, Pacer, candidates
+from zot.fetch.resolve import BROWSER_UA, Notes, Pacer, candidates
 from zot.fetch.select import ItemRef
 
 STATS = (
@@ -26,6 +26,7 @@ STATS = (
     "no_candidate",
     "all_failed",
     "quota",
+    "error",
 )
 _UNSAFE = re.compile(r"[^A-Za-z0-9 ._-]+")
 
@@ -57,6 +58,103 @@ def _scratch_path(scratch_dir: Path, ref: ItemRef, cand_key: str, ext: str) -> P
     return scratch_dir / f"{ref.key}-{tag}.{ext}"
 
 
+def _cleanup(paths: list[Path]) -> None:
+    for p in paths:
+        p.unlink(missing_ok=True)
+
+
+def _one(
+    db: Db,
+    ref: ItemRef,
+    *,
+    notes: Notes,
+    scratch: list[Path],
+    budget: Budget,
+    clients: Clients,
+    email: str,
+    storage_dir: Path,
+    scratch_dir: Path,
+    dry_run: bool,
+    copier: Copier,
+    sleep: Callable[[float], None],
+    out: Callable[..., None],
+    up_pacer: Pacer,
+    lg_pacer: Pacer,
+    ledger: Ledger,
+) -> str:
+    """One item end to end; returns the stats key for its outcome
+    ("" for a dry run). Raises on unexpected failure — ``run`` isolates it."""
+    cands = candidates(
+        ref,
+        direct=clients.direct,
+        proxied=clients.proxied,
+        email=email,
+        up_pacer=up_pacer,
+        lg_pacer=lg_pacer,
+        notes=notes,
+    )
+    if dry_run:
+        out(f"{ref.key}  {ref.title[:60]}  ids={ref.ids}")
+        for c in cands:
+            out(f"    {c.source:<9} {c.ext:<4} {c.md5 or c.url}")
+        if not cands:
+            out(f"    (no candidates: {notes})")
+        return ""
+    if not cands:
+        outcome = "error" if notes.had_error else "no_candidate"
+        ledger.record(ref.key, outcome=outcome, ids=ref.ids, note=str(notes))
+        out(f"{ref.key}  {outcome}  {ref.title[:60]}")
+        return outcome
+
+    budget.quota_hit = False
+    delivered: tuple[str, Path, str] | None = None
+    for c in cands:
+        dest = _scratch_path(scratch_dir, ref, c.md5 or c.url or "", c.ext)
+        scratch.append(dest)
+        source = fetch(
+            c,
+            dest,
+            direct=clients.direct,
+            proxied=clients.proxied,
+            budget=budget,
+            pacer=lg_pacer,
+            sleep=sleep,
+            notes=notes,
+        )
+        if source:
+            delivered = (source, dest, c.md5 or "")
+            break
+    if delivered is None:
+        if budget.quota_hit:
+            outcome = "quota"
+        elif notes.had_error:
+            outcome = "error"
+        else:
+            outcome = "all_failed"
+        ledger.record(ref.key, outcome=outcome, ids=ref.ids, note=str(notes))
+        out(f"{ref.key}  {outcome}  {ref.title[:60]}")
+        return outcome
+
+    source, path, md5 = delivered
+    final = path.with_name(f"{safe_name(ref.title, ref.key)}{path.suffix}")
+    path.replace(final)
+    scratch.append(final)
+    attach_file(
+        db,
+        ref.item_id,
+        final,
+        source=source,
+        md5=md5,
+        storage_dir=storage_dir,
+        copier=copier,
+    )
+    db.commit()
+    _cleanup(scratch)
+    ledger.record(ref.key, outcome="attached", source=source, md5=md5, ids=ref.ids)
+    out(f"{ref.key}  attached  {source}  {final.name}")
+    return "attached"
+
+
 def run(
     db: Db,
     refs: list[ItemRef],
@@ -87,65 +185,39 @@ def run(
             stats["skipped"] += 1
             continue
         stats["attempted"] += 1
-        cands = candidates(
-            ref,
-            direct=clients.direct,
-            proxied=clients.proxied,
-            email=email,
-            up_pacer=up_pacer,
-            lg_pacer=lg_pacer,
-        )
-        if dry_run:
-            out(f"{ref.key}  {ref.title[:60]}  ids={ref.ids}")
-            for c in cands:
-                out(f"    {c.source:<9} {c.ext:<4} {c.md5 or c.url}")
-            if not cands:
-                out("    (no candidates)")
-            continue
-        if not cands:
-            ledger.record(ref.key, outcome="no_candidate", ids=ref.ids)
-            stats["no_candidate"] += 1
-            out(f"{ref.key}  no_candidate  {ref.title[:60]}")
-            continue
-
-        budget.quota_hit = False
-        delivered: tuple[str, Path, str] | None = None
-        for c in cands:
-            dest = _scratch_path(scratch_dir, ref, c.md5 or c.url or "", c.ext)
-            source = fetch(
-                c,
-                dest,
-                direct=clients.direct,
-                proxied=clients.proxied,
+        notes = Notes()
+        scratch: list[Path] = []
+        try:
+            outcome = _one(
+                db,
+                ref,
+                notes=notes,
+                scratch=scratch,
                 budget=budget,
-                pacer=lg_pacer,
+                clients=clients,
+                email=email,
+                storage_dir=storage_dir,
+                scratch_dir=scratch_dir,
+                dry_run=dry_run,
+                copier=copier,
                 sleep=sleep,
+                out=out,
+                up_pacer=up_pacer,
+                lg_pacer=lg_pacer,
+                ledger=ledger,
             )
-            if source:
-                delivered = (source, dest, c.md5 or "")
-                break
-        if delivered is None:
-            outcome = "quota" if budget.quota_hit else "all_failed"
-            ledger.record(ref.key, outcome=outcome, ids=ref.ids)
-            stats[outcome] += 1
-            out(f"{ref.key}  {outcome}  {ref.title[:60]}")
+        except AnnasAuthError:
+            db.con.rollback()
+            _cleanup(scratch)
+            raise
+        except Exception as exc:  # per-item failures never abort the run
+            db.con.rollback()
+            _cleanup(scratch)
+            note = f"{type(exc).__name__}: {exc}"
+            ledger.record(ref.key, outcome="error", ids=ref.ids, note=note)
+            stats["error"] += 1
+            out(f"{ref.key}  error  {note[:120]}")
             continue
-
-        source, path, md5 = delivered
-        final = path.with_name(f"{safe_name(ref.title, ref.key)}{path.suffix}")
-        path.replace(final)
-        attach_file(
-            db,
-            ref.item_id,
-            final,
-            source=source,
-            md5=md5,
-            storage_dir=storage_dir,
-            copier=copier,
-        )
-        db.commit()
-        final.unlink(missing_ok=True)
-        ledger.record(ref.key, outcome="attached", source=source, md5=md5, ids=ref.ids)
-        stats["attached"] += 1
-        out(f"{ref.key}  attached  {source}  {final.name}")
+        if outcome:
+            stats[outcome] += 1
     return stats

tests/zot/fetch/test_attach.py

@@ -127,3 +127,26 @@ def test_sudo_copier_runs_mkdir_cp_chown(monkeypatch, tmp_path: Path):
         ["sudo", "cp", str(tmp_path / "a.pdf"), str(dest / "a.pdf")],
         ["sudo", "chown", "-R", "100999:100999", str(dest)],
     ]
+
+
+def test_attach_file_writes_rows_before_copying(zdb: Db, tmp_path: Path):
+    """A failed copy must leave rows the caller can roll back, never an
+    orphan storage dir with no row."""
+    parent = mk_item(zdb, title="P")
+    src = tmp_path / "P.pdf"
+    src.write_bytes(b"%PDF-1.4 x")
+    seen = {}
+
+    def spy_copier(s: Path, d: Path) -> None:
+        seen["rows"] = zdb.con.execute(
+            "SELECT COUNT(*) FROM itemAttachments WHERE parentItemID = ?", (parent,)
+        ).fetchone()[0]
+        raise OSError("disk full")
+
+    with pytest.raises(OSError):
+        attach_file(
+            zdb, parent, src, source="libgen", storage_dir=tmp_path, copier=spy_copier
+        )
+    assert seen["rows"] == 1
+    zdb.con.rollback()
+    assert has_full_text(zdb, parent) is False

tests/zot/fetch/test_deliver.py

@@ -14,7 +14,7 @@ from zot.fetch.deliver import (
     fetch_annas,
     fetch_libgen,
 )
-from zot.fetch.resolve import LIBGEN, Candidate, Pacer
+from zot.fetch.resolve import LIBGEN, Candidate, Notes, Pacer
 
 FX = Path(__file__).parent / "fixtures"
 PDF = b"%PDF-1.4\n" + b"0" * MIN_BYTES
@@ -339,3 +339,91 @@ def test_fetch_all_fail_without_annas_is_not_quota(tmp_path):
         sleep=_nosleep,
     )
     assert src is None and b.quota_hit is False
+
+
+# ── fix pass (review) ─────────────────────────────────────────────
+
+
+def test_fetch_annas_quota_403_disables_instead_of_raising(tmp_path):
+    b = Budget(annas_key="SECRET")
+    api = httpx.Response(403, json={"download_url": None, "error": "No downloads left"})
+    notes = Notes()
+    got = fetch_annas(
+        _client(_annas_handler(api)), _lg(), tmp_path / "f.pdf", budget=b, notes=notes
+    )
+    assert got is None
+    assert b.annas_disabled is True and b.annas_left == 0
+    assert notes.had_error and "No downloads left" in str(notes)
+
+
+@pytest.mark.parametrize("err", ["Invalid secret key", "Not a member"])
+def test_fetch_annas_key_errors_still_raise(tmp_path, err):
+    b = Budget(annas_key="SECRET")
+    api = httpx.Response(403, json={"download_url": None, "error": err})
+    with pytest.raises(AnnasAuthError, match=err):
+        fetch_annas(_client(_annas_handler(api)), _lg(), tmp_path / "f.pdf", budget=b)
+
+
+def test_notes_distinguish_errors_from_misses(tmp_path):
+    # transport error → error note; definitive 404 → miss note
+    def boom(r):
+        raise httpx.ConnectError("x")
+
+    n = Notes()
+    fetch(
+        _lg(),
+        tmp_path / "f.pdf",
+        direct=_client(boom),
+        proxied=_client(boom),
+        budget=Budget(),
+        pacer=_pacer(),
+        sleep=_nosleep,
+        notes=n,
+    )
+    assert n.had_error and "ConnectError" in str(n)
+
+    n2 = Notes()
+    fetch(
+        _up(),
+        tmp_path / "f.pdf",
+        direct=_client(lambda r: httpx.Response(404)),
+        proxied=_client(boom),
+        budget=Budget(),
+        pacer=_pacer(),
+        notes=n2,
+    )
+    assert not n2.had_error and "404" in str(n2)
+
+
+def test_libgen_throttle_page_is_an_error_note(tmp_path):
+    err = httpx.Response(200, text='<div class="alert alert-danger">db 3306</div>')
+    h, _ = _libgen_handler([err, err])
+    n = Notes()
+    fetch_libgen(
+        _client(h), _lg(), tmp_path / "f.pdf", pacer=_pacer(), sleep=_nosleep, notes=n
+    )
+    assert n.had_error and "ads" in str(n)
+
+
+def test_libgen_and_annas_never_use_the_direct_client(tmp_path):
+    def never(r):
+        raise AssertionError("direct client used for libgen/Anna's")
+
+    def h(r: httpx.Request) -> httpx.Response:
+        if r.url.path == "/ads.php":
+            return httpx.Response(500)
+        if r.url.host == "annas-archive.gl":
+            return httpx.Response(200, text=ANNAS_OK)
+        return httpx.Response(200, content=PDF, headers=PDF_HDR)
+
+    b = Budget(annas_key="SECRET")
+    src = fetch(
+        _lg(),
+        tmp_path / "f.pdf",
+        direct=_client(never),
+        proxied=_client(h),
+        budget=b,
+        pacer=_pacer(),
+        sleep=_nosleep,
+    )
+    assert src == "annas"

tests/zot/fetch/test_ledger.py

@@ -99,3 +99,9 @@ def test_default_now_is_utc_iso(tmp_path: Path):
     led = Ledger(tmp_path / "ledger.jsonl")
     att = led.record("K1", outcome="attached")
     assert att.ts.endswith("+00:00")
+
+
+def test_error_outcome_retries_immediately(tmp_path: Path):
+    led = Ledger(tmp_path / "ledger.jsonl")
+    led.record("K1", outcome="error", note="ConnectError: proxy down", now=T0)
+    assert led.should_try("K1", now=T0) is True

tests/zot/fetch/test_resolve.py

@@ -10,6 +10,7 @@ from zot.fetch.resolve import (
     BROWSER_UA,
     LIBGEN,
     Candidate,
+    Notes,
     Pacer,
     candidates,
     libgen_candidates,
@@ -267,3 +268,66 @@ def test_libgen_candidates_skips_non_dict_editions():
     body = json.dumps({"1": None, "2": [1, 2], "3": {"title": "T", "files": []}})
     c = _client(lambda req: httpx.Response(200, text=body))
     assert libgen_candidates(c, kind="paper", doi="10.1/x", pacer=_pacer()) == []
+
+
+# ── fix pass (review) ─────────────────────────────────────────────
+
+
+def test_notes_error_for_http_and_transport_miss_for_empty():
+    n = Notes()
+    libgen_editions(
+        _client(lambda r: httpx.Response(503)), doi="10.1/x", pacer=_pacer(), notes=n
+    )
+    assert n.had_error and "503" in str(n)
+
+    def boom(req):
+        raise httpx.ConnectError("down")
+
+    n2 = Notes()
+    libgen_editions(_client(boom), doi="10.1/x", pacer=_pacer(), notes=n2)
+    assert n2.had_error and "ConnectError" in str(n2)
+
+    n3 = Notes()
+    libgen_candidates(
+        _client(lambda r: httpx.Response(200, text="[]")),
+        kind="paper",
+        doi="10.1/x",
+        pacer=_pacer(),
+        notes=n3,
+    )
+    assert not n3.had_error and "no edition" in str(n3)
+
+
+def test_candidates_note_unpaywall_miss():
+    n = Notes()
+    got = candidates(
+        _ref(doi="10.1/x"),
+        direct=_direct(False),
+        proxied=_client(lambda r: httpx.Response(200, text="[]")),
+        email="dev@fhirworx.io",
+        up_pacer=_pacer(),
+        lg_pacer=_pacer(),
+        notes=n,
+    )
+    assert got == [] and "unpaywall" in str(n) and not n.had_error
+
+
+def test_libgen_candidates_tolerates_bad_file_entries():
+    body = json.dumps(
+        {
+            "1": {
+                "title": "T",
+                "files": {"a": None, "b": 3, "c": {"f_id": "9", "md5": "c" * 32}},
+            }
+        }
+    )
+
+    def h(r):
+        if r.url.params.get("object") == "f":
+            return httpx.Response(
+                200, json={"9": {"extension": "pdf", "filesize": "n/a"}}
+            )
+        return httpx.Response(200, text=body)
+
+    got = libgen_candidates(_client(h), kind="paper", doi="10.1/x", pacer=_pacer())
+    assert [(g.md5, g.ext, g.size_bytes) for g in got] == [("c" * 32, "pdf", None)]

tests/zot/fetch/test_run.py

@@ -64,13 +64,16 @@ def _libgen_none(r: httpx.Request) -> httpx.Response:
 
 
 def _libgen_dead(r: httpx.Request) -> httpx.Response:
+    """Every route answers definitively (a miss), nothing errors."""
     if r.url.path == "/json.php" and r.url.params.get("object") == "e":
         return httpx.Response(200, text=(FX / "libgen_json_doi.json").read_text())
     if r.url.path == "/json.php":
         return httpx.Response(200, text=(FX / "libgen_json_file.json").read_text())
+    if r.url.path == "/ads.php":
+        return httpx.Response(200, text=ADS)
     if r.url.host == "annas-archive.gl":
         return httpx.Response(200, json={"download_url": None, "error": "nope"})
-    return httpx.Response(500)
+    return httpx.Response(404)
 
 
 def _setup(tmp_path: Path, **kw):
@@ -80,7 +83,7 @@ def _setup(tmp_path: Path, **kw):
         email="dev@fhirworx.io",
         storage_dir=tmp_path / "storage",
         scratch_dir=tmp_path / "scratch",
-        copier=_py_copier,
+        copier=kw.pop("copier", _py_copier),
         sleep=lambda s: None,
         out=kw.pop("out", lambda *a: None),
         **kw,
@@ -210,3 +213,90 @@ def test_run_annas_auth_error_aborts(zdb: Db, tmp_path: Path):
     refs = select_items(zdb, all_=True)
     with pytest.raises(AnnasAuthError):
         run(zdb, refs, clients=_clients(_unpaywall_miss, proxied), **kw)
+
+
+# ── fix pass (review) ─────────────────────────────────────────────
+
+
+def _proxy_down(r: httpx.Request) -> httpx.Response:
+    raise httpx.ConnectError("proxy down")
+
+
+def test_run_transient_failure_is_error_not_terminal(zdb: Db, tmp_path: Path):
+    mk_item(zdb, title="T", doi=DOI)
+    kw = _setup(tmp_path)
+    refs = select_items(zdb, all_=True)
+    stats = run(zdb, refs, clients=_clients(_unpaywall_miss, _proxy_down), **kw)
+    assert stats == _zero(seen=1, attempted=1, error=1)
+    led = kw["ledger"]
+    assert led.summary() == {"error": 1}
+    assert led.should_try(refs[0].key) is True  # retried next run, not 30 days out
+    line = (tmp_path / "ledger.jsonl").read_text()
+    assert "ConnectError" in line
+
+
+def test_run_all_failed_and_no_candidate_carry_notes(zdb: Db, tmp_path: Path):
+    mk_item(zdb, title="no ids")
+    mk_item(zdb, title="dead", doi=DOI)
+    kw = _setup(tmp_path)
+    refs = select_items(zdb, all_=True)
+    run(zdb, refs, clients=_clients(_unpaywall_miss, _libgen_dead), **kw)
+    lines = (tmp_path / "ledger.jsonl").read_text().splitlines()
+    assert "no identifiers" in lines[0]
+    assert "unpaywall" in lines[1] and "libgen get: http 404" in lines[1]
+
+
+def test_run_isolates_a_failing_item_and_rolls_back(zdb: Db, tmp_path: Path):
+    bad = mk_item(zdb, title="bad", doi=DOI)
+    good = mk_item(zdb, title="good", doi=DOI)
+    calls = {"n": 0}
+
+    def flaky_copier(src: Path, dest_dir: Path) -> None:
+        calls["n"] += 1
+        if calls["n"] == 1:
+            raise RuntimeError("sudo: a password is required")
+        _py_copier(src, dest_dir)
+
+    kw = _setup(tmp_path, copier=flaky_copier)
+    refs = select_items(zdb, all_=True)
+    stats = run(zdb, refs, clients=_clients(_unpaywall_ok, _libgen_none), **kw)
+    assert stats == _zero(seen=2, attempted=2, attached=1, error=1)
+    # the failed item left no rows behind (rolled back) and no scratch file
+    rows = zdb.con.execute(
+        "SELECT COUNT(*) FROM itemAttachments WHERE parentItemID = ?", (bad,)
+    ).fetchone()[0]
+    assert rows == 0 and "source:unpaywall" not in zdb.get_tags(bad)
+    assert has_full_text(zdb, good)
+    assert not list((tmp_path / "scratch").iterdir())
+    assert "RuntimeError" in (tmp_path / "ledger.jsonl").read_text()
+    assert kw["ledger"].should_try(refs[0].key) is True
+
+
+def test_run_annas_auth_error_still_aborts_after_fix(zdb: Db, tmp_path: Path):
+    mk_item(zdb, title="T", doi=DOI)
+
+    def proxied(r: httpx.Request) -> httpx.Response:
+        if r.url.host == "annas-archive.gl":
+            return httpx.Response(403, json={"error": "Invalid secret key"})
+        return _libgen_dead(r)
+
+    kw = _setup(tmp_path, budget=Budget(annas_key="bad"))
+    refs = select_items(zdb, all_=True)
+    with pytest.raises(AnnasAuthError):
+        run(zdb, refs, clients=_clients(_unpaywall_miss, proxied), **kw)
+
+
+def test_run_delivery_error_is_error_outcome(zdb: Db, tmp_path: Path):
+    """Candidates exist but every delivery route errored (5xx) → retry next run."""
+
+    def flaky(r: httpx.Request) -> httpx.Response:
+        if r.url.path == "/ads.php":
+            return httpx.Response(503)
+        return _libgen_ok(r)
+
+    mk_item(zdb, title="T", doi=DOI)
+    kw = _setup(tmp_path)
+    refs = select_items(zdb, all_=True)
+    stats = run(zdb, refs, clients=_clients(_unpaywall_miss, flaky), **kw)
+    assert stats == _zero(seen=1, attempted=1, error=1)
+    assert "503" in (tmp_path / "ledger.jsonl").read_text()