diff --git a/application/cmd/cre_main.py b/application/cmd/cre_main.py index b3dfe91c4..216f7c7b6 100644 --- a/application/cmd/cre_main.py +++ b/application/cmd/cre_main.py @@ -38,6 +38,60 @@ app = None +def fetch_upstream_json( + path: str, + timeout: Optional[float] = None, + max_attempts: Optional[int] = None, + backoff_seconds: Optional[float] = None, +) -> Dict[str, Any]: + base_url = os.environ.get("CRE_UPSTREAM_API_URL", "https://opencre.org/rest/v1") + timeout = timeout or float(os.environ.get("CRE_UPSTREAM_TIMEOUT_SECONDS", "30")) + max_attempts = max_attempts or int(os.environ.get("CRE_UPSTREAM_MAX_ATTEMPTS", "4")) + backoff_seconds = backoff_seconds or float( + os.environ.get("CRE_UPSTREAM_RETRY_BACKOFF_SECONDS", "2") + ) + url = f"{base_url}{path}" + last_error: Optional[Exception] = None + + for attempt in range(1, max_attempts + 1): + response = None + try: + response = requests.get(url, timeout=timeout) + if response.status_code == 200: + return response.json() + + status_error = RuntimeError( + f"cannot connect to upstream status code {response.status_code}" + ) + if response.status_code < 500 and response.status_code != 429: + raise status_error + last_error = status_error + except requests.exceptions.RequestException as exc: + last_error = exc + + if attempt < max_attempts: + retry_after_header = None + sleep_seconds = backoff_seconds * attempt + if response is not None and response.status_code == 429: + retry_after_header = response.headers.get("Retry-After") + try: + if retry_after_header is not None: + sleep_seconds = float(retry_after_header) + except (TypeError, ValueError): + sleep_seconds = backoff_seconds * attempt + logger.warning( + "upstream fetch failed for %s on attempt %s/%s, retrying", + url, + attempt, + max_attempts, + ) + time.sleep(sleep_seconds) + + if last_error: + raise RuntimeError(f"upstream fetch failed for {url}") from last_error + raise RuntimeError(f"upstream fetch failed for {url}") + + def register_node(node: defs.Node, collection: db.Node_collection) -> db.Node: """ for each link find if either the root node or the link have a CRE, @@ -591,19 +645,11 @@ def download_graph_from_upstream(cache: str) -> None: collection = db_connect(path=cache).with_graph() def download_cre_from_upstream(creid: str): - cre_response = requests.get( - os.environ.get("CRE_UPSTREAM_API_URL", "https://opencre.org/rest/v1") - + f"/id/{creid}" - ) - if cre_response.status_code != 200: - raise RuntimeError( - f"cannot connect to upstream status code {cre_response.status_code}" - ) - data = cre_response.json() + if creid in imported_cres: + return + data = fetch_upstream_json(f"/id/{creid}") credict = data["data"] cre = defs.Document.from_dict(credict) - if cre.id in imported_cres: - return register_cre(cre, collection) imported_cres[cre.id] = "" @@ -611,15 +657,7 @@ def download_cre_from_upstream(creid: str): if link.document.doctype == defs.Credoctypes.CRE: download_cre_from_upstream(link.document.id) - root_cres_response = requests.get( - os.environ.get("CRE_UPSTREAM_API_URL", "https://opencre.org/rest/v1") - + "/root_cres" - ) - if root_cres_response.status_code != 200: - raise RuntimeError( - f"cannot connect to upstream status code {root_cres_response.status_code}" - ) - data = root_cres_response.json() + data = fetch_upstream_json("/root_cres") for root_cre in data["data"]: cre = defs.Document.from_dict(root_cre) register_cre(cre, collection) diff --git a/application/tests/cre_main_test.py b/application/tests/cre_main_test.py index 097b0b6d9..61bb23c9e 100644 --- a/application/tests/cre_main_test.py +++ b/application/tests/cre_main_test.py @@ -6,6 +6,7 @@ from typing import Any, Dict, List from unittest import mock from unittest.mock import Mock, patch +import requests from rq import Queue, job from application.utils import redis from application.prompt_client import prompt_client as prompt_client @@ -470,6 +471,75 @@ def test_register_cre(self) -> None: ], ) + @patch("application.cmd.cre_main.time.sleep") + @patch("application.cmd.cre_main.requests.get") + def test_fetch_upstream_json_retries_transient_failures( + self, mock_get, mock_sleep + ) -> None: + transient_error = requests.exceptions.ConnectionError("reset by peer") + success_response = Mock() + success_response.status_code = 200 + success_response.json.return_value = {"data": []} + mock_get.side_effect = [transient_error, success_response] + + # Make retry behaviour deterministic for the test by passing explicit + # retry settings instead of relying on environment defaults. + data = main.fetch_upstream_json("/root_cres", max_attempts=2, backoff_seconds=1) + + self.assertEqual(data, {"data": []}) + self.assertEqual(mock_get.call_count, 2) + mock_sleep.assert_called_once() + + @patch("application.cmd.cre_main.time.sleep") + @patch("application.cmd.cre_main.requests.get") + def test_fetch_upstream_json_fails_fast_on_non_retryable_status( + self, mock_get, mock_sleep + ) -> None: + response = Mock() + response.status_code = 404 + response.headers = {} + mock_get.return_value = response + + with self.assertRaises(RuntimeError): + main.fetch_upstream_json("/root_cres") + + self.assertEqual(mock_get.call_count, 1) + mock_sleep.assert_not_called() + + @patch("application.cmd.cre_main.time.sleep") + @patch("application.cmd.cre_main.requests.get") + def test_fetch_upstream_json_retries_retryable_status_until_exhausted( + self, mock_get, mock_sleep + ) -> None: + response = Mock() + response.status_code = 503 + response.headers = {} + mock_get.return_value = response + + with self.assertRaises(RuntimeError): + main.fetch_upstream_json("/root_cres", max_attempts=3, backoff_seconds=2) + + self.assertEqual(mock_get.call_count, 3) + self.assertEqual(mock_sleep.call_count, 2) + mock_sleep.assert_any_call(2) + mock_sleep.assert_any_call(4) + + @patch("application.cmd.cre_main.time.sleep") + @patch("application.cmd.cre_main.requests.get") + def test_fetch_upstream_json_honors_retry_after_for_rate_limit( + self, mock_get, mock_sleep + ) -> None: + response = Mock() + response.status_code = 429 + response.headers = {"Retry-After": "7"} + mock_get.side_effect = [response, response, RuntimeError("should not reach")] + + with self.assertRaises(RuntimeError): + main.fetch_upstream_json("/root_cres", max_attempts=2, backoff_seconds=2) + + self.assertEqual(mock_get.call_count, 2) + mock_sleep.assert_called_once_with(7.0) + @patch.object(main, "db_connect") @patch.object(Queue, "enqueue_call") @patch.object(redis, "wait_for_jobs")