fix(harvester): recover records skipped by live pagination

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,129 @@ 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, *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 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 +211,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)

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"]
+    )