From 2f637d7e60d90fcc86b9e6886680126632f8d252 Mon Sep 17 00:00:00 2001 From: Mohammed Taha Khan Date: Wed, 29 Jul 2026 16:08:25 +0200 Subject: [PATCH] fix(harvester): recover records skipped by live pagination --- site/cds_rdm/inspire_harvester/reader.py | 152 ++++++++++++++---- .../inspire_harvester/test_harvester_job.py | 118 ++++++++++++++ site/tests/inspire_harvester/test_reader.py | 136 ++++++++++++++++ 3 files changed, 377 insertions(+), 29 deletions(-) diff --git a/site/cds_rdm/inspire_harvester/reader.py b/site/cds_rdm/inspire_harvester/reader.py index 22c2d346..3aeab48c 100644 --- a/site/cds_rdm/inspire_harvester/reader.py +++ b/site/cds_rdm/inspire_harvester/reader.py @@ -15,6 +15,8 @@ from cds_rdm.inspire_harvester.transform.resource_types import ALL_DOCUMENT_TYPES +INSPIRE_LITERATURE_API = "https://inspirehep.net/api/literature" + class InspireHTTPReader(BaseReader): """INSPIRE HTTP Reader.""" @@ -40,40 +42,134 @@ def __init__( super().__init__(origin, mode, *args, **kwargs) - def _iter(self, url, *args, **kwargs): + def _build_url(self, q, **params): + """Build an INSPIRE literature search URL.""" + query_params = {"q": q, **params} + return f"{INSPIRE_LITERATURE_API}?{urlencode(query_params)}" + + def _get_json(self, url, headers): + """Fetch JSON from INSPIRE or raise ReaderError.""" + current_app.logger.info(f"Querying URL: {url}.") + response = requests.get(url, headers=headers) + if response.status_code != 200: + error_message = ( + f"Error occurred while getting JSON data from INSPIRE. " + f"See URL: {url}. Error message: {response.text}. " + f"Status code: {response.status_code}" + ) + current_app.logger.error(error_message) + raise ReaderError(error_message) + current_app.logger.debug("Request response is successful (200).") + return response.json() + + def _scan_ids(self, q, headers): + """Paginate an ID-only search and return IDs plus the reported total.""" + ids = set() + url = self._build_url(q, fields="id", size=1000) + reported_total = None + + while url: + data = self._get_json(url, headers) + if reported_total is None: + reported_total = data["hits"]["total"] + for hit in data["hits"]["hits"]: + ids.add(str(hit["id"])) + url = data.get("links", {}).get("next") + + return ids, reported_total + + def _iter(self, url, q=None, *args, **kwargs): """Yields HTTP response.""" # header set to include additional data (external file URLs and more detailed metadata headers = {"Accept": "application/vnd+inspire.record.expanded+json"} initial_url = url + seen_ids = set() + expected_total = None + had_another_page = False while url: # Continue until there is no "next" link - current_app.logger.info(f"Querying URL: {url}.") - response = requests.get(url, headers=headers) - data = response.json() - if response.status_code == 200: - current_app.logger.debug("Request response is successful (200).") - total = data["hits"]["total"] - hits = data["hits"]["hits"] - - if total == 0: - current_app.logger.warning( - f"No results found when querying INSPIRE. See URL: {url}." - ) - elif url == initial_url: - current_app.logger.info(f"Records found: {total}.") - - for inspire_record in hits: - current_app.logger.debug( - f"Sending INSPIRE record #{inspire_record['id']} to transformer." - ) - yield inspire_record - else: - error_message = f"Error occurred while getting JSON data from INSPIRE. See URL: {url}. Error message: {response.text}. Status code: {response.status_code}" - current_app.logger.error(error_message) - raise ReaderError(error_message) + data = self._get_json(url, headers) + total = data["hits"]["total"] + hits = data["hits"]["hits"] + + if total == 0: + current_app.logger.warning( + f"No results found when querying INSPIRE. See URL: {url}." + ) + elif url == initial_url: + expected_total = total + current_app.logger.info(f"Records found: {total}.") + + for inspire_record in hits: + record_id = str(inspire_record["id"]) + if record_id in seen_ids: + continue + seen_ids.add(record_id) + current_app.logger.debug( + f"Sending INSPIRE record #{record_id} to transformer." + ) + yield inspire_record # Get the next page URL if available url = data.get("links", {}).get("next") + # Remember if there was a second page. + if url: + had_another_page = True + + # If results moved while we were paging, we may have missed some records. + # Skip this check for single-id jobs, or when everything fit on one page. + if ( + not q + or expected_total is None + or self._inspire_id + or not had_another_page + ): + return + + # One ID-only scan, then harvest anything we missed. + # Retry only if the scan collected fewer IDs than INSPIRE reported. + all_ids, scan_total = self._scan_ids(q, headers) + if scan_total is not None and len(all_ids) != scan_total: + current_app.logger.warning( + "ID scan found fewer INSPIRE records than reported. " + f"| details: found={len(all_ids)}, reported={scan_total}" + ) + retry_ids, _ = self._scan_ids(q, headers) + if retry_ids != all_ids: + current_app.logger.warning( + "INSPIRE ID scan shifted on retry. " + f"| details: first={len(all_ids)}, retry={len(retry_ids)}" + ) + all_ids |= retry_ids + + # IDs we still need to fetch (in INSPIRE's list, not in what we already got). + missing_ids = all_ids - seen_ids + if not missing_ids: + return + + current_app.logger.info( + "Re-fetching missing INSPIRE records. " + f"| details: missing={len(missing_ids)}, missing_ids={missing_ids}" + ) + for record_id in missing_ids: + data = self._get_json( + self._build_url(f"{q} AND id:{record_id}"), + headers, + ) + for inspire_record in data["hits"]["hits"]: + recovered_id = str(inspire_record["id"]) + seen_ids.add(recovered_id) + current_app.logger.debug( + f"Sending INSPIRE record #{recovered_id} to transformer." + ) + yield inspire_record + + still_missing = all_ids - seen_ids + if still_missing: + current_app.logger.warning( + "After recovery, some INSPIRE records are still missing. " + f"| details: missing_ids={still_missing}" + ) def read(self, item=None, *args, **kwargs): """Builds a query depending on the input data.""" @@ -120,11 +216,9 @@ def read(self, item=None, *args, **kwargs): ) query_params = {"q": f"{q} AND du >= {self._since}"} - base_url = "https://inspirehep.net/api/literature" - encoded_query = urlencode(query_params) - url = f"{base_url}?{encoded_query}" + url = self._build_url(query_params["q"]) current_app.logger.info( f"Resulting query: {query_params['q']}. URL for harvesting data from INSPIRE: {url}." ) - yield from self._iter(url=url, *args, **kwargs) + yield from self._iter(url=url, q=query_params["q"], *args, **kwargs) diff --git a/site/tests/inspire_harvester/test_harvester_job.py b/site/tests/inspire_harvester/test_harvester_job.py index ad17050e..fd60eade 100644 --- a/site/tests/inspire_harvester/test_harvester_job.py +++ b/site/tests/inspire_harvester/test_harvester_job.py @@ -441,3 +441,121 @@ def mock_requests_get_pagination( tranformation(created_record2.to_dict()["hits"]["hits"][0]["id"], expected_result_2) tranformation(created_record3.to_dict()["hits"]["hits"][0]["id"], expected_result_3) + + +def test_inspire_job_recovers_pagination_shift(running_app, scientific_community, caplog): + """Full harvest persists a record skipped by mid-pagination INSPIRE shifts.""" + page_1_file = DATA_DIR / "inspire_response_15_records_page_1.json" + page_2_file = DATA_DIR / "inspire_response_15_records_page_2.json" + with open(page_1_file) as f: + page_1_data = json.load(f) + with open(page_2_file) as f: + page_2_data = json.load(f) + + by_id = { + str(hit["id"]): hit + for hit in page_1_data["hits"]["hits"] + page_2_data["hits"]["hits"] + } + # Known-good thesis fixtures from test_inspire_job. + record_a = by_id["2802969"] + record_b = by_id["1452604"] + record_c = by_id["2840463"] # skipped in first pass + skipped_id = str(record_c["id"]) + + page_1 = { + "hits": {"total": 3, "hits": [record_a, record_b]}, + "links": { + "next": ( + "https://inspirehep.net/api/literature" + "?q=_oai.sets%3AForCDS+AND+du+%3E%3D+2024-11-15+AND+du+%3C%3D+2025-01-09" + "&size=2&page=2" + ) + }, + } + # After a live update, record C moved to page 1; page 2 no longer has it. + page_2_shifted = { + "hits": {"total": 3, "hits": []}, + "links": {}, + } + ids_page = { + "hits": { + "total": 3, + "hits": [ + {"id": str(record_a["id"])}, + {"id": str(record_b["id"])}, + {"id": skipped_id}, + ], + }, + "links": {}, + } + missing_page = { + "hits": {"total": 1, "hits": [record_c]}, + "links": {}, + } + + ds_config = { + "config": { + "readers": [ + { + "type": "inspire-http-reader", + "args": { + "since": "2024-11-15", + "until": "2025-01-09", + }, + }, + ], + "transformers": [{"type": "inspire-json-transformer"}], + "writers": [ + { + "type": "async", + "args": { + "writer": { + "type": "inspire-writer", + } + }, + } + ], + "batch_size": 100, + "write_many": True, + } + } + + def mock_requests_get_shift( + url, + headers={"Accept": "application/vnd+inspire.record.expanded+json"}, + stream=True, + ): + if "fields=id" in url: + content = ids_page + elif f"id%3A{skipped_id}" in url or f"id:{skipped_id}" in url: + content = missing_page + elif "page=2" in url: + content = page_2_shifted + else: + content = page_1 + return mock_requests_get(url, mock_content=content) + + run_harvester_mock(ds_config, mock_requests_get_shift) + + RDMRecord.index.refresh() + + assert "Re-fetching missing INSPIRE records." in caplog.text + assert "missing=1" in caplog.text + assert skipped_id in caplog.text + + for inspire_id in (str(record_a["id"]), str(record_b["id"]), skipped_id): + created = current_rdm_records_service.search( + system_identity, + params={"q": f"metadata.related_identifiers.identifier:{inspire_id}"}, + ) + assert created.total == 1, f"Expected CDS record for INSPIRE#{inspire_id}" + + # Explicitly prove the shifted record was persisted. + skipped_record = current_rdm_records_service.search( + system_identity, + params={"q": f"metadata.related_identifiers.identifier:{skipped_id}"}, + ) + assert ( + skipped_record.to_dict()["hits"]["hits"][0]["metadata"]["title"] + == record_c["metadata"]["titles"][0]["title"] + ) diff --git a/site/tests/inspire_harvester/test_reader.py b/site/tests/inspire_harvester/test_reader.py index 70676872..a7e4d2b0 100644 --- a/site/tests/inspire_harvester/test_reader.py +++ b/site/tests/inspire_harvester/test_reader.py @@ -62,3 +62,139 @@ def test_reader_success(running_app): assert "metadata" in data assert "id" in data assert "links" in data + + +def test_reader_recovers_missing_records(running_app, caplog): + """Test reader recovers records skipped by mid-harvest pagination shifts.""" + page_1 = { + "hits": { + "total": 6, + "hits": [{"id": "1"}, {"id": "2"}, {"id": "3"}], + }, + "links": { + "next": "https://inspirehep.net/api/literature?q=test&page=2", + }, + } + # Record 4 moved to page 1 after a live update, so page 2 no longer has it. + page_2 = { + "hits": { + "total": 6, + "hits": [{"id": "5"}, {"id": "6"}], + }, + "links": {}, + } + ids_page = { + "hits": { + "total": 6, + "hits": [{"id": str(i)} for i in range(1, 7)], + }, + "links": {}, + } + missing_record = { + "hits": { + "total": 1, + "hits": [{"id": "4", "metadata": {"titles": [{"title": "Missing"}]}}], + }, + "links": {}, + } + + def side_effect(url, headers=None): + mock_response = Mock() + mock_response.status_code = 200 + if "fields=id" in url: + mock_response.json.return_value = ids_page + elif "id%3A4" in url or "id:4" in url: + mock_response.json.return_value = missing_record + elif "page=2" in url: + mock_response.json.return_value = page_2 + else: + mock_response.json.return_value = page_1 + return mock_response + + with patch("requests.get", side_effect=side_effect): + reader = InspireHTTPReader(since="2024-01-01", until="2024-01-02") + records = list(reader.read()) + + assert [str(r["id"]) for r in records] == ["1", "2", "3", "5", "6", "4"] + assert "Re-fetching missing INSPIRE records." in caplog.text + assert "missing=1" in caplog.text + assert "4" in caplog.text + + +def test_reader_recovers_when_count_matches(running_app, caplog): + """Test reader recovers when counts match but query IDs differ.""" + page_1 = { + "hits": { + "total": 6, + "hits": [{"id": "1"}, {"id": "2"}, {"id": "3"}], + }, + "links": { + "next": "https://inspirehep.net/api/literature?q=test&page=2", + }, + } + page_2 = { + "hits": { + "total": 6, + "hits": [{"id": "4"}, {"id": "5"}, {"id": "6"}], + }, + "links": {}, + } + ids_page = { + "hits": { + "total": 6, + "hits": [{"id": str(i)} for i in (1, 2, 3, 5, 6, 7)], + }, + "links": {}, + } + missing_record = { + "hits": { + "total": 1, + "hits": [{"id": "7", "metadata": {"titles": [{"title": "New"}]}}], + }, + "links": {}, + } + + def side_effect(url, headers=None): + mock_response = Mock() + mock_response.status_code = 200 + if "fields=id" in url: + mock_response.json.return_value = ids_page + elif "id%3A7" in url or "id:7" in url: + mock_response.json.return_value = missing_record + elif "page=2" in url: + mock_response.json.return_value = page_2 + else: + mock_response.json.return_value = page_1 + return mock_response + + with patch("requests.get", side_effect=side_effect): + reader = InspireHTTPReader(since="2024-01-01", until="2024-01-02") + records = list(reader.read()) + + assert [str(r["id"]) for r in records] == ["1", "2", "3", "4", "5", "6", "7"] + assert "Re-fetching missing INSPIRE records." in caplog.text + + +def test_reader_skips_recovery_for_single_page(running_app, caplog): + """Single-page harvests skip the ID scan (no pagination skip risk).""" + page_1 = { + "hits": { + "total": 2, + "hits": [{"id": "1"}, {"id": "2"}], + }, + "links": {}, + } + + def side_effect(url, headers=None): + mock_response = Mock() + mock_response.status_code = 200 + mock_response.json.return_value = page_1 + return mock_response + + with patch("requests.get", side_effect=side_effect) as mock_get: + reader = InspireHTTPReader(since="2024-01-01", until="2024-01-02") + records = list(reader.read()) + + assert [str(r["id"]) for r in records] == ["1", "2"] + assert "Re-fetching missing INSPIRE records." not in caplog.text + assert all("fields=id" not in call.args[0] for call in mock_get.call_args_list) \ No newline at end of file