From 622d44a215608383beb1ed3fb9a3217396abb50c Mon Sep 17 00:00:00 2001 From: PRAteek-singHWY Date: Wed, 29 Apr 2026 23:31:39 +0530 Subject: [PATCH 1/3] Fix: Prevent infinite loop in gap_analysis preload (#885) --- application/utils/gap_analysis.py | 29 ++++++++++++++++++++--------- 1 file changed, 20 insertions(+), 9 deletions(-) diff --git a/application/utils/gap_analysis.py b/application/utils/gap_analysis.py index 9e3dab04d..7a1bcdf8e 100644 --- a/application/utils/gap_analysis.py +++ b/application/utils/gap_analysis.py @@ -198,14 +198,25 @@ def calculate_a_to_b(sa: str, sb: str) -> bool: for sb in standards: if sa == sb: continue - waiting.append(f"{sa}->{sb}") + waiting.append((sa, sb)) + + MAX_RETRIES = 10 + retries = {pair: 0 for pair in waiting} + while len(waiting): - for sa in standards: - for sb in standards: - if sa == sb: - continue - if calculate_a_to_b(sa, sb): - waiting.remove(f"{sa}->{sb}") if f"{sa}->{sb}" in waiting else "" - print(f"calculating {len(waiting)} gap analyses") - time.sleep(30) + for pair in list(waiting): + sa, sb = pair + if calculate_a_to_b(sa, sb): + waiting.remove(pair) + else: + retries[pair] += 1 + if retries[pair] >= MAX_RETRIES: + print( + f"{sa}->{sb} reached max retries ({MAX_RETRIES}), dropping from waiting list" + ) + waiting.remove(pair) + + if waiting: + print(f"calculating {len(waiting)} gap analyses") + time.sleep(30) print("map analysis preloaded successfully") From 5f948c458aa7725788708cbb15071f7c2567644f Mon Sep 17 00:00:00 2001 From: PRAteek-singHWY Date: Thu, 5 Nov 2026 16:36:41 +0530 Subject: [PATCH 2/3] gap analysis preload: only count real failures against the retry cap A pair still running under the same job id is no longer a failure, so long GA jobs are not dropped after 10 polls. A changed job id counts as one, because map_analysis only enqueues a new job once the last one ended without a result. preload now reports dropped pairs and no longer logs success when any were dropped. Adds tests. --- .../tests/gap_analysis_preload_test.py | 94 +++++++++++++++++++ application/utils/gap_analysis.py | 85 ++++++++++++----- 2 files changed, 157 insertions(+), 22 deletions(-) create mode 100644 application/tests/gap_analysis_preload_test.py diff --git a/application/tests/gap_analysis_preload_test.py b/application/tests/gap_analysis_preload_test.py new file mode 100644 index 000000000..33a8e8da0 --- /dev/null +++ b/application/tests/gap_analysis_preload_test.py @@ -0,0 +1,94 @@ +import unittest +from typing import Any, Callable, Dict, List +from unittest.mock import MagicMock, patch + +from application.utils import gap_analysis + + +def _response(status: int = 200, body: Any = None) -> MagicMock: + res = MagicMock() + res.status_code = status + res.json.return_value = body if body is not None else {} + return res + + +class PreloadTest(unittest.TestCase): + """preload() polls map_analysis until every pair is cached or given up on.""" + + def _run( + self, standards: List[str], answer: Callable[[str, int], MagicMock] + ) -> Dict[str, Any]: + polls: Dict[str, int] = {} + + def fake_get(url: str) -> MagicMock: + if url.endswith("/rest/v1/standards"): + return _response(body=standards) + pair = url.split("standard=", 1)[1].replace("&standard=", "->") + polls[pair] = polls.get(pair, 0) + 1 + return answer(pair, polls[pair]) + + with patch.object(gap_analysis.requests, "get", side_effect=fake_get), patch( + "application.utils.gap_analysis.time.sleep" + ) as sleep: + dropped = gap_analysis.preload(target_url="http://x") + return {"dropped": dropped, "polls": polls, "sleeps": sleep.call_count} + + def test_all_pairs_cached_needs_one_pass_and_no_sleep(self) -> None: + out = self._run(["A", "B"], lambda _p, _n: _response(body={"result": {"k": 1}})) + self.assertEqual([], out["dropped"]) + self.assertEqual({"A->B": 1, "B->A": 1}, out["polls"]) + self.assertEqual(0, out["sleeps"]) + + def test_finished_pair_is_not_polled_again(self) -> None: + def answer(pair: str, n: int) -> MagicMock: + if pair == "A->B" or n >= 3: + return _response(body={"result": {"k": 1}}) + return _response(body={"job_id": "job-1"}) + + out = self._run(["A", "B"], answer) + self.assertEqual(1, out["polls"]["A->B"]) + self.assertEqual(3, out["polls"]["B->A"]) + + def test_long_running_job_is_not_dropped(self) -> None: + # GA jobs may run for 36 hours, so polling the same job id is not a failure. + polls_while_running = gap_analysis.PRELOAD_MAX_FAILURES * 5 + + def answer(_pair: str, n: int) -> MagicMock: + if n > polls_while_running: + return _response(body={"result": {"k": 1}}) + return _response(body={"job_id": "same-job"}) + + out = self._run(["A", "B"], answer) + self.assertEqual([], out["dropped"]) + + def test_pair_that_keeps_returning_404_is_dropped(self) -> None: + out = self._run(["A", "B"], lambda _p, _n: _response(status=404)) + self.assertEqual(["A->B", "B->A"], sorted(out["dropped"])) + self.assertEqual(gap_analysis.PRELOAD_MAX_FAILURES, out["polls"]["A->B"]) + + def test_job_that_keeps_failing_and_requeueing_is_dropped(self) -> None: + # map_analysis enqueues a new job once the last one failed, so a changed + # job id means the previous job ended without producing a result. + out = self._run(["A", "B"], lambda p, n: _response(body={"job_id": f"{p}{n}"})) + self.assertEqual(["A->B", "B->A"], sorted(out["dropped"])) + # The first poll only records the job id, it is not a failure. + self.assertEqual(gap_analysis.PRELOAD_MAX_FAILURES + 1, out["polls"]["A->B"]) + + def test_failures_are_counted_per_pair(self) -> None: + def answer(pair: str, _n: int) -> MagicMock: + if pair == "A->B": + return _response(status=404) + return _response(body={"result": {"k": 1}}) + + out = self._run(["A", "B"], answer) + self.assertEqual(["A->B"], out["dropped"]) + + def test_success_is_not_logged_when_pairs_were_dropped(self) -> None: + with self.assertLogs(gap_analysis.logger, level="INFO") as logs: + self._run(["A", "B"], lambda _p, _n: _response(status=404)) + self.assertFalse(any("preloaded successfully" in m for m in logs.output)) + self.assertTrue(any("A->B" in m and "ERROR" in m for m in logs.output)) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/utils/gap_analysis.py b/application/utils/gap_analysis.py index 49810a278..7bf850855 100644 --- a/application/utils/gap_analysis.py +++ b/application/utils/gap_analysis.py @@ -5,7 +5,7 @@ import os import requests import time -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, Tuple from application.utils import redis from flask import json as flask_json import json @@ -22,6 +22,8 @@ } GAP_ANALYSIS_TIMEOUT = "129600s" # 36 hours +PRELOAD_MAX_FAILURES = 10 +PRELOAD_POLL_SECONDS = 30 OPENCRE_STANDARD_NAME = "OpenCRE" OPENCRE_OVERLAP_LINK_TYPES = ( defs.LinkTypes.LinkedTo, @@ -400,42 +402,81 @@ def schedule(standards: List[str], database): return {"result": res.get("result") if isinstance(res, dict) else res} -def preload(target_url: str): - waiting = [] +def preload(target_url: str) -> List[str]: + """ + Poll map_analysis until every standard pair is cached or given up on. + + A pair still running under the same job id is left alone, since GA jobs may + run for GAP_ANALYSIS_TIMEOUT. A pair counts a failure on a non-200 response, + on a 200 with neither result nor job id, and when its job id changes, because + map_analysis only enqueues a new job once the last one ended without a result. + After PRELOAD_MAX_FAILURES a pair is dropped so one bad pair cannot block the + run forever. + + Returns the dropped pairs as ``A->B`` strings. + """ standards_request = requests.get(f"{target_url}/rest/v1/standards") standards = standards_request.json() - def calculate_a_to_b(sa: str, sb: str) -> bool: + def poll_pair(sa: str, sb: str) -> Tuple[bool, Optional[str]]: + """Returns (cached, job_id).""" res = requests.get( f"{target_url}/rest/v1/map_analysis?standard={sa}&standard={sb}" ) if res.status_code != 200: logger.warning("%s->%s returned %s", sa, sb, res.status_code) - return False + return False, None tojson = res.json() if tojson.get("result"): - return True + return True, None if tojson.get("job_id"): logger.info("%s->%s waiting", sa, sb) - return False + return False, tojson["job_id"] logger.warning( "%s->%s returned 200 but has no 'result' or 'job_id' key", sa, sb ) - return False - - for sa in standards: - for sb in standards: - if sa == sb: + return False, None + + waiting = [(sa, sb) for sa in standards for sb in standards if sa != sb] + failures: Dict[Tuple[str, str], int] = {pair: 0 for pair in waiting} + last_job_ids: Dict[Tuple[str, str], str] = {} + dropped: List[str] = [] + + while waiting: + for pair in list(waiting): + cached, job_id = poll_pair(*pair) + if cached: + waiting.remove(pair) continue - waiting.append(f"{sa}->{sb}") - while len(waiting): - for sa in standards: - for sb in standards: - if sa == sb: + + previous_job_id = last_job_ids.get(pair) + if job_id: + last_job_ids[pair] = job_id + if previous_job_id is None or previous_job_id == job_id: continue - if calculate_a_to_b(sa, sb): - waiting.remove(f"{sa}->{sb}") if f"{sa}->{sb}" in waiting else "" - logger.info("calculating %s gap analyses", len(waiting)) - time.sleep(30) - logger.info("map analysis preloaded successfully") + + failures[pair] += 1 + if failures[pair] >= PRELOAD_MAX_FAILURES: + logger.warning( + "%s->%s failed %s times, dropping it", + pair[0], + pair[1], + failures[pair], + ) + waiting.remove(pair) + dropped.append(f"{pair[0]}->{pair[1]}") + + if waiting: + logger.info("calculating %s gap analyses", len(waiting)) + time.sleep(PRELOAD_POLL_SECONDS) + + if dropped: + logger.error( + "map analysis preload gave up on %s pairs: %s", + len(dropped), + ", ".join(dropped), + ) + else: + logger.info("map analysis preloaded successfully") + return dropped From f631ed48e3c994fe19f48da906b62cc69d02111c Mon Sep 17 00:00:00 2001 From: PRAteek-singHWY Date: Thu, 5 Nov 2026 17:05:54 +0530 Subject: [PATCH 3/3] gap analysis preload: a job still running clears earlier failures Ten scattered 503s could otherwise drop a healthy long job. The cap now means failures in a row, and the docstring says a new job also follows an expired in-flight key. --- application/tests/gap_analysis_preload_test.py | 11 +++++++++++ application/utils/gap_analysis.py | 10 +++++++--- 2 files changed, 18 insertions(+), 3 deletions(-) diff --git a/application/tests/gap_analysis_preload_test.py b/application/tests/gap_analysis_preload_test.py index 33a8e8da0..bb1ce7d4f 100644 --- a/application/tests/gap_analysis_preload_test.py +++ b/application/tests/gap_analysis_preload_test.py @@ -74,6 +74,17 @@ def test_job_that_keeps_failing_and_requeueing_is_dropped(self) -> None: # The first poll only records the job id, it is not a failure. self.assertEqual(gap_analysis.PRELOAD_MAX_FAILURES + 1, out["polls"]["A->B"]) + def test_scattered_errors_do_not_drop_a_running_job(self) -> None: + def answer(_pair: str, n: int) -> MagicMock: + if n > 200: + return _response(body={"result": {"k": 1}}) + if n % 20 == 0: + return _response(status=503) + return _response(body={"job_id": "same-job"}) + + out = self._run(["A", "B"], answer) + self.assertEqual([], out["dropped"]) + def test_failures_are_counted_per_pair(self) -> None: def answer(pair: str, _n: int) -> MagicMock: if pair == "A->B": diff --git a/application/utils/gap_analysis.py b/application/utils/gap_analysis.py index 7bf850855..a6ce54dd5 100644 --- a/application/utils/gap_analysis.py +++ b/application/utils/gap_analysis.py @@ -409,9 +409,10 @@ def preload(target_url: str) -> List[str]: A pair still running under the same job id is left alone, since GA jobs may run for GAP_ANALYSIS_TIMEOUT. A pair counts a failure on a non-200 response, on a 200 with neither result nor job id, and when its job id changes, because - map_analysis only enqueues a new job once the last one ended without a result. - After PRELOAD_MAX_FAILURES a pair is dropped so one bad pair cannot block the - run forever. + map_analysis enqueues a new job when the last one ended without a result (or + its in-flight key expired). Seeing the same job again clears the count. After + PRELOAD_MAX_FAILURES failures in a row a pair is dropped, so one bad pair + cannot block the run forever. Returns the dropped pairs as ``A->B`` strings. """ @@ -454,6 +455,9 @@ def poll_pair(sa: str, sb: str) -> Tuple[bool, Optional[str]]: if job_id: last_job_ids[pair] = job_id if previous_job_id is None or previous_job_id == job_id: + # A job that is still alive clears earlier blips, so ten + # scattered 503s cannot drop a healthy long-running job. + failures[pair] = 0 continue failures[pair] += 1