Fix: Prevent infinite loop in gap_analysis preload (#885)

application/tests/gap_analysis_preload_test.py

@@ -0,0 +1,105 @@
+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_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":
+                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()

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,85 @@ 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 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.
+    """
     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:
+                    # A job that is still alive clears earlier blips, so ten
+                    # scattered 503s cannot drop a healthy long-running job.
+                    failures[pair] = 0
                     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