diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index f4472ec..15d2bec 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -338,8 +338,9 @@ it will be re-fetched and rewritten **Cause:** -This is deliberate, and it is a one-off per reference. Cache entries record -which extractor wrote them (`extractor_version` in the file's frontmatter). +This is deliberate; the migration completes for each reference only when its +refresh succeeds. Cache entries record which extractor wrote them +(`extractor_version` in the file's frontmatter). Versions before the extractor fixes discarded some full-text articles and cached a short PMC placeholder in their place, labelled as full text, and welded text across inline markup — so entries written then hold content that @@ -375,6 +376,25 @@ for the reasons above. The entry is left stale rather than rewritten, so the next run that can reach the source refreshes it properly. This warning is printed without `-v`. +Until a refresh succeeds, each new run attempts to fetch every distinct stale +reference before falling back to its cached text. Repeated uses of the same ID +share an in-memory result, but this does not cover other IDs or later processes. +Offline runs with large caches may therefore spend substantial time waiting for +network failures on every run; there is no circuit breaker. The wait depends on +the source's timeout and retry policy. + +For PubMed, Bio.Entrez handles HTTP/URL errors while opening requests (three +attempts by default, with its own delays). The validator does not restart an +exhausted Bio.Entrez retry loop. Dropped connections, socket/TLS errors, and +HTTP framing errors (including incomplete bodies) are attempted up to three +times, with 2 and 4 second +backoff between attempts. Each attempt opens a new request, so mixed opening +and body failures can involve up to nine HTTP requests with Bio.Entrez defaults +per endpoint (summary and article XML). Exhaustion reports the reference as +unfetchable, or uses an eligible stale cached copy if one is available. A partial +summary refresh does not replace useful cached text. These are attempt limits, +not a wall-clock deadline. + ### Title validation failed **Symptom:** diff --git a/src/linkml_reference_validator/etl/sources/pmid.py b/src/linkml_reference_validator/etl/sources/pmid.py index d954b68..99a2177 100644 --- a/src/linkml_reference_validator/etl/sources/pmid.py +++ b/src/linkml_reference_validator/etl/sources/pmid.py @@ -10,10 +10,14 @@ True """ +from collections.abc import Callable +from http.client import HTTPException +from io import BytesIO import logging import re import time from typing import Any, Optional +from urllib.error import HTTPError, URLError from Bio import Entrez # type: ignore from bs4 import BeautifulSoup # type: ignore @@ -108,15 +112,18 @@ def fetch( pmid = identifier.strip() Entrez.email = config.email # type: ignore - time.sleep(config.rate_limit_delay) + summary = self._read_entrez( + lambda: Entrez.esummary(db="pubmed", id=pmid), pmid, config + ) + if summary is None: + return None - # External API call - handle network/API errors + # Parse only complete responses: a truncated stream can otherwise look + # like malformed XML to Entrez.read instead of a transport failure. try: - handle = Entrez.esummary(db="pubmed", id=pmid) - records = Entrez.read(handle) - handle.close() - except Exception as e: - logger.warning(f"Failed to fetch PMID:{pmid} from NCBI: {e}") + records = Entrez.read(BytesIO(summary)) + except (ValueError, RuntimeError) as exc: + logger.warning("Failed to parse PMID:%s summary: %s", pmid, exc) return None if not records: @@ -143,6 +150,10 @@ def fetch( # A single efetch of the article XML backs the abstract, MeSH terms, # and publication types, so we don't round-trip to NCBI three times. article_xml = self._fetch_pubmed_xml(pmid, config) + if article_xml is None: + # A failed refresh must not overwrite useful cached text with just + # summary metadata. None also enables ReferenceFetcher's stale fallback. + return None abstract = self._parse_abstract(article_xml) if article_xml else None keywords = self._parse_mesh_terms(article_xml) if article_xml else None publication_types = ( @@ -235,6 +246,55 @@ def _parse_abstract(self, soup: BeautifulSoup) -> Optional[str]: joined = "\n\n".join(sections) return joined if joined else None + def _read_entrez( + self, + open_handle: Callable[[], Any], + pmid: str, + config: ReferenceValidationConfig, + ) -> Optional[bytes]: + """Read a complete response, retrying transport failures up to three attempts. + + Entrez already retries HTTP/URL errors during opening (three attempts by + default, with its own delays). Do not restart that exhausted retry loop. + Socket/TLS failures and HTTP framing errors escape Entrez's loop, so + retry those here with 2 and 4 second backoff. A mixed sequence can make + at most ``3 * Entrez.max_tries`` requests; no process-global Entrez retry + settings are changed. Parsing happens after this transport-only boundary. + + Args: + open_handle: Open a new Entrez response for each attempt. + pmid: PubMed identifier for diagnostics. + config: Configuration for rate limiting. + + Returns: + Complete response bytes, or None when the request fails. + """ + for attempt in range(3): + time.sleep(config.rate_limit_delay) + handle = None + try: + try: + handle = open_handle() + except URLError as exc: + # Includes HTTPError. Entrez has already applied its retry + # policy; retrying here would multiply outage waits. + if isinstance(exc, HTTPError): + exc.close() + logger.warning("Failed to open PMID:%s from NCBI: %s", pmid, exc) + return None + return handle.read() + except (OSError, HTTPException) as exc: + logger.warning( + "NCBI transport failure for PMID:%s (attempt %s/3): %s", + pmid, attempt + 1, exc, + ) + finally: + if handle is not None: + handle.close() + if attempt < 2: + time.sleep(2 ** (attempt + 1)) + return None + def _fetch_pubmed_xml( self, pmid: str, config: ReferenceValidationConfig ) -> Optional[BeautifulSoup]: @@ -250,12 +310,11 @@ def _fetch_pubmed_xml( Returns: Parsed BeautifulSoup document, or None if nothing was returned """ - time.sleep(config.rate_limit_delay) - - handle = Entrez.efetch(db="pubmed", id=pmid, - rettype="xml", retmode="xml") - xml_content = handle.read() - handle.close() + xml_content = self._read_entrez( + lambda: Entrez.efetch(db="pubmed", id=pmid, rettype="xml", retmode="xml"), + pmid, + config, + ) if not xml_content: return None diff --git a/tests/test_pmid_network.py b/tests/test_pmid_network.py new file mode 100644 index 0000000..264f3e2 --- /dev/null +++ b/tests/test_pmid_network.py @@ -0,0 +1,327 @@ +"""Exercise live Entrez transport and parsing against a controlled HTTP server.""" + +from collections import Counter +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from ssl import SSLEOFError +from threading import Event, Thread +from urllib.parse import urlsplit +from urllib.request import Request, urlopen + +from Bio import Entrez +from Bio.Entrez import Parser +import pytest + +from linkml_reference_validator.etl.reference_fetcher import ( + EXTRACTOR_CACHE_VERSION, + ReferenceFetcher, +) +from linkml_reference_validator.etl.sources.pmid import PMIDSource +from linkml_reference_validator.models import ( + ReferenceContent, + ReferenceValidationConfig, +) +from linkml_reference_validator.validation.supporting_text_validator import ( + SupportingTextValidator, +) + +SUMMARY = b""" + +123A study +Smith J +""" +ARTICLE = b"The patient recovered." + + +@pytest.fixture +def ncbi(monkeypatch): + """Redirect real Entrez urlopen to HTTP responses, including truncated bodies.""" + failures = {} + counts = Counter() + handles = [] + sleeps = [] + + class Handler(BaseHTTPRequestHandler): + """Serve scripted failures before a valid Entrez response.""" + + def do_GET(self): + """Drop headers/body or send a complete XML response.""" + endpoint = urlsplit(self.path).path.rsplit("/", 1)[-1] + counts[endpoint] += 1 + script = failures.get(endpoint, []) + action = script.pop(0) if script else "ok" + if action == "timeout": + Event().wait(3) + self.close_connection = True + return + if action == "bad_status": + self.wfile.write(b"Not an HTTP status\r\n\r\n") + self.close_connection = True + return + if action == "disconnect": + self.close_connection = True + return + if isinstance(action, int): + self.send_error(action) + return + payload = SUMMARY if endpoint == "esummary.fcgi" else ARTICLE + if action == "empty": + payload = b"" + if action == "invalid": + payload = b"not XML" + if action == "no_abstract": + payload = b"" + self.send_response(200) + self.send_header("Content-Type", "text/xml") + self.send_header("X-Test-Failure", action) + self.send_header("Content-Length", str(len(payload))) + self.end_headers() + if action == "read_timeout": + Event().wait(3) + self.close_connection = True + return + self.wfile.write(payload[:20] if action == "truncate" else payload) + self.close_connection = True + + def log_message(self, format, *args): + """Suppress expected server error logging.""" + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + thread = Thread(target=server.serve_forever, daemon=True) + thread.start() + + class TLSFailureHandle: + """Inject a TLS read failure after consuming bytes from a real HTTP handle. + + This boundary harness avoids relying on platform-specific OpenSSL EOF + suppression, while exercising real request creation, IO and cleanup. + """ + + def __init__(self, handle): + self.handle = handle + + def __getattr__(self, name): + return getattr(self.handle, name) + + def read(self): + """Consume a partial response before surfacing the TLS exception.""" + assert self.handle.read(20) + raise SSLEOFError("TLS connection closed during response body") + + def local_urlopen(request): + """Keep Entrez request creation/retries and use a real local HTTP handle.""" + parts = urlsplit(request.full_url) + local = Request( + f"http://127.0.0.1:{server.server_port}{parts.path}?{parts.query}", + data=request.data, + headers=dict(request.header_items()), + ) + handle = urlopen(local, timeout=1) + handles.append(handle) + if handle.headers.get("X-Test-Failure") == "tls_eof": + return TLSFailureHandle(handle) + return handle + + def forbid_dtd_network(*args, **kwargs): + """Ensure summary parsing uses the DTD bundled with Biopython.""" + pytest.fail("Entrez parser attempted an external DTD request") + + monkeypatch.setattr(Parser, "urlopen", forbid_dtd_network) + monkeypatch.setattr(Entrez, "urlopen", local_urlopen) + # Entrez.time is stdlib time: this also records validator sleeps. The + # HTTP handler uses Event.wait so its intended timeouts remain real. + monkeypatch.setattr(Entrez.time, "sleep", sleeps.append) + # Use the real built-in retry loop with its normal limit. + monkeypatch.setattr(Entrez, "max_tries", 3) + try: + yield failures, counts, handles, sleeps + finally: + server.shutdown() + server.server_close() + thread.join() + + +@pytest.fixture +def config(tmp_path): + """Isolate cache writes and disable unrelated full-text providers.""" + return ReferenceValidationConfig( + cache_dir=tmp_path, fetch_full_text=False, rate_limit_delay=0 + ) + + +@pytest.mark.parametrize("endpoint", ["esummary.fcgi", "efetch.fcgi"]) +@pytest.mark.parametrize( + "failure", + ["truncate", "disconnect", "timeout", "read_timeout", "bad_status", "tls_eof"], +) +def test_transient_recovery(ncbi, config, endpoint, failure): + """Opening and body failures recover, and every response handle closes.""" + failures, counts, handles, sleeps = ncbi + failures[endpoint] = [failure, failure] + ref = PMIDSource().fetch("123", config) + assert ref is not None + assert ref.content == "The patient recovered." + assert counts[endpoint] == 3 + assert all(handle.closed for handle in handles) + assert 2 in sleeps and 4 in sleeps + + +@pytest.mark.parametrize("endpoint", ["esummary.fcgi", "efetch.fcgi"]) +@pytest.mark.parametrize( + "failure", + ["truncate", "disconnect", "timeout", "read_timeout", "bad_status", "tls_eof"], +) +def test_exhaustion_does_not_pass_or_abort_next_reference( + ncbi, config, endpoint, failure +): + """An exhausted PMID fails validation while the next PMID still succeeds.""" + failures, counts, handles, _ = ncbi + failures[endpoint] = [failure] * 3 + validator = SupportingTextValidator(config) + assert not validator.validate("The patient recovered.", "PMID:123").is_valid + assert counts[endpoint] == 3 + assert validator.validate("The patient recovered.", "PMID:456").is_valid + assert all(handle.closed for handle in handles) + + +@pytest.mark.parametrize("status,attempts", [(400, 1), (429, 3), (503, 3)]) +@pytest.mark.parametrize("endpoint", ["esummary.fcgi", "efetch.fcgi"]) +def test_entrez_open_retries_are_not_multiplied( + ncbi, config, endpoint, status, attempts +): + """Already-exhausted Entrez HTTP retries are not wrapped in another loop.""" + failures, counts, handles, _ = ncbi + failures[endpoint] = [status] * 12 + assert PMIDSource().fetch("123", config) is None + assert counts[endpoint] == attempts + assert all(handle.closed for handle in handles) + + +def test_summary_parse_error_is_not_retried(ncbi, config): + """A complete but invalid summary is a deterministic failure, not a retry.""" + failures, counts, handles, _ = ncbi + failures["esummary.fcgi"] = ["invalid"] + assert PMIDSource().fetch("123", config) is None + assert counts["esummary.fcgi"] == 1 + assert all(handle.closed for handle in handles) + + +@pytest.mark.parametrize("force_refresh", [False, True]) +def test_partial_refresh_preserves_stale_cache(ncbi, config, force_refresh): + """Summary success plus XML exhaustion must not replace useful stale text.""" + fetcher = ReferenceFetcher(config) + old = ReferenceContent( + reference_id="PMID:123", + content="Useful older text.", + content_type="abstract_only", + ) + fetcher._save_to_disk(old) + path = fetcher.get_cache_path("PMID:123") + stale = path.read_text().replace( + f"extractor_version: {EXTRACTOR_CACHE_VERSION}", "extractor_version: 0" + ) + path.write_text(stale) + failures, counts, _, _ = ncbi + failures["efetch.fcgi"] = ["truncate"] * 3 + ref = fetcher.fetch("PMID:123", force_refresh=force_refresh) + assert counts["efetch.fcgi"] == 3 + if force_refresh: + assert ref is None + else: + assert ref is not None and ref.content == old.content + assert fetcher.fetch("PMID:123") is ref + assert counts["efetch.fcgi"] == 3 + assert path.read_text() == stale + + +def test_article_parser_is_outside_retry_boundary(ncbi, config, monkeypatch): + """A parser defect after successful real HTTP fetches is raised just once.""" + _, counts, handles, _ = ncbi + calls = [] + + def broken_parser(content, parser): + """Represent a deterministic extraction defect after transport succeeds.""" + calls.append(content) + raise ValueError("invalid parser configuration") + + monkeypatch.setattr( + "linkml_reference_validator.etl.sources.pmid.BeautifulSoup", broken_parser + ) + with pytest.raises(ValueError, match="invalid parser configuration"): + PMIDSource().fetch("123", config) + assert calls == [ARTICLE] + assert counts["efetch.fcgi"] == 1 + assert all(handle.closed for handle in handles) + + +def test_complete_record_without_abstract_remains_unavailable(ncbi, config): + """A successful no-abstract response remains distinct from transport failure.""" + failures, counts, _, _ = ncbi + failures["efetch.fcgi"] = ["no_abstract"] + ref = PMIDSource().fetch("123", config) + assert ref is not None + assert ref.title == "A study" + assert ref.content_type == "unavailable" + assert ref.content is None + assert counts["efetch.fcgi"] == 1 + + +@pytest.mark.parametrize("endpoint", ["esummary.fcgi", "efetch.fcgi"]) +@pytest.mark.parametrize("last_response", ["ok", "truncate"]) +def test_mixed_open_and_body_failures_have_a_finite_bound( + ncbi, config, endpoint, last_response +): + """Built-in open retries followed by body failures take at most nine requests.""" + failures, counts, handles, _ = ncbi + failures[endpoint] = [503, 503, "truncate"] * 2 + [503, 503, last_response] + ref = PMIDSource().fetch("123", config) + assert (ref is not None) == (last_response == "ok") + assert counts[endpoint] == 9 + assert Entrez.max_tries == 3 + assert all(handle.closed for handle in handles) + + +def test_connection_refused_uses_only_entrez_retries(ncbi, config, monkeypatch): + """Real connection refusal exhausts the library limit without outer retries.""" + import socket + + _, _, _, sleeps = ncbi + attempts = [] + with socket.socket() as endpoint: + # A bound socket without listen() reserves a port that refuses connections. + endpoint.bind(("127.0.0.1", 0)) + port = endpoint.getsockname()[1] + + def refused_urlopen(request): + """Exercise urllib's real connection-error wrapping.""" + attempts.append(request) + return urlopen(f"http://127.0.0.1:{port}/", timeout=1) + + monkeypatch.setattr(Entrez, "urlopen", refused_urlopen) + assert PMIDSource().fetch("123", config) is None + assert len(attempts) == 3 + assert sleeps.count(Entrez.sleep_between_tries) == 2 + assert 2 not in sleeps and 4 not in sleeps + + +def test_empty_article_response_does_not_replace_stale_text(ncbi, config): + """A complete empty body provides no replacement article text for a cache.""" + fetcher = ReferenceFetcher(config) + fetcher._save_to_disk( + ReferenceContent( + reference_id="PMID:123", + content="Useful older text.", + content_type="abstract_only", + ) + ) + path = fetcher.get_cache_path("PMID:123") + stale = path.read_text().replace( + f"extractor_version: {EXTRACTOR_CACHE_VERSION}", "extractor_version: 0" + ) + path.write_text(stale) + failures, counts, handles, _ = ncbi + failures["efetch.fcgi"] = ["empty"] + ref = fetcher.fetch("PMID:123") + assert ref is not None and ref.content == "Useful older text." + assert path.read_text() == stale + assert counts["efetch.fcgi"] == 1 + assert all(handle.closed for handle in handles) diff --git a/tests/test_sources.py b/tests/test_sources.py index 2a50d27..a4d886e 100644 --- a/tests/test_sources.py +++ b/tests/test_sources.py @@ -437,6 +437,7 @@ def test_fetch_uses_single_efetch( self, mock_read, mock_esummary, mock_efetch, source, config ): """fetch() should derive abstract, MeSH, and pub types from one efetch.""" + mock_esummary.return_value.read.return_value = b"" mock_read.return_value = [ { "Title": "An illustrative case",