From 7d2373beae1a947e436d6df457f001e1aca76596 Mon Sep 17 00:00:00 2001 From: bornunique911 Date: Wed, 1 Jul 2026 18:06:56 +0530 Subject: [PATCH 1/4] feat: retry transient failures during upstream sync (#471) --- application/cmd/cre_main.py | 64 +++++++++++++++++++++--------- application/tests/cre_main_test.py | 18 +++++++++ 2 files changed, 64 insertions(+), 18 deletions(-) diff --git a/application/cmd/cre_main.py b/application/cmd/cre_main.py index b3dfe91c4..e42b61689 100644 --- a/application/cmd/cre_main.py +++ b/application/cmd/cre_main.py @@ -38,6 +38,50 @@ 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): + 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: + logger.warning( + "upstream fetch failed for %s on attempt %s/%s, retrying", + url, + attempt, + max_attempts, + ) + time.sleep(backoff_seconds * attempt) + + 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,15 +635,7 @@ 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() + data = fetch_upstream_json(f"/id/{creid}") credict = data["data"] cre = defs.Document.from_dict(credict) if cre.id in imported_cres: @@ -611,15 +647,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..947325a29 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,23 @@ 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] + + data = main.fetch_upstream_json("/root_cres") + + self.assertEqual(data, {"data": []}) + self.assertEqual(mock_get.call_count, 2) + mock_sleep.assert_called_once() + @patch.object(main, "db_connect") @patch.object(Queue, "enqueue_call") @patch.object(redis, "wait_for_jobs") From a112bc3fc8971c8c410df44c7178df33a26894d5 Mon Sep 17 00:00:00 2001 From: bornunique911 Date: Wed, 1 Jul 2026 18:18:29 +0530 Subject: [PATCH 2/4] feat: enhance upstream fetch with retry logic for rate limits and non-retryable errors --- application/cmd/cre_main.py | 15 +++++++-- application/tests/cre_main_test.py | 50 ++++++++++++++++++++++++++++++ 2 files changed, 62 insertions(+), 3 deletions(-) diff --git a/application/cmd/cre_main.py b/application/cmd/cre_main.py index e42b61689..807efe53a 100644 --- a/application/cmd/cre_main.py +++ b/application/cmd/cre_main.py @@ -69,13 +69,22 @@ def fetch_upstream_json( last_error = exc if attempt < max_attempts: + retry_after_header = None + sleep_seconds = backoff_seconds * attempt + if "response" in locals() 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(backoff_seconds * attempt) + time.sleep(sleep_seconds) if last_error: raise RuntimeError(f"upstream fetch failed for {url}") from last_error @@ -635,11 +644,11 @@ def download_graph_from_upstream(cache: str) -> None: collection = db_connect(path=cache).with_graph() def download_cre_from_upstream(creid: str): + 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] = "" diff --git a/application/tests/cre_main_test.py b/application/tests/cre_main_test.py index 947325a29..66b60facc 100644 --- a/application/tests/cre_main_test.py +++ b/application/tests/cre_main_test.py @@ -488,6 +488,56 @@ def test_fetch_upstream_json_retries_transient_failures( 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") From 43ae711d70e07dd7f84dfaaa18263de257f00e23 Mon Sep 17 00:00:00 2001 From: Bornunique911 <69379200+Bornunique911@users.noreply.github.com> Date: Wed, 1 Jul 2026 21:40:53 +0530 Subject: [PATCH 3/4] Update application/cmd/cre_main.py Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com> Signed-off-by: Bornunique911 <69379200+Bornunique911@users.noreply.github.com> --- application/cmd/cre_main.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/application/cmd/cre_main.py b/application/cmd/cre_main.py index 807efe53a..216f7c7b6 100644 --- a/application/cmd/cre_main.py +++ b/application/cmd/cre_main.py @@ -54,6 +54,7 @@ def fetch_upstream_json( 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: @@ -71,7 +72,7 @@ def fetch_upstream_json( if attempt < max_attempts: retry_after_header = None sleep_seconds = backoff_seconds * attempt - if "response" in locals() and response.status_code == 429: + 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: From a1fa64ef006c34ebba3f630a51b353857cf982aa Mon Sep 17 00:00:00 2001 From: bornunique911 Date: Wed, 1 Jul 2026 21:51:04 +0530 Subject: [PATCH 4/4] test: make retry behavior deterministic in upstream fetch test --- application/tests/cre_main_test.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/application/tests/cre_main_test.py b/application/tests/cre_main_test.py index 66b60facc..61bb23c9e 100644 --- a/application/tests/cre_main_test.py +++ b/application/tests/cre_main_test.py @@ -482,7 +482,9 @@ def test_fetch_upstream_json_retries_transient_failures( success_response.json.return_value = {"data": []} mock_get.side_effect = [transient_error, success_response] - data = main.fetch_upstream_json("/root_cres") + # 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)