diff --git a/.env.example b/.env.example index 17ad95d27..a7345bb7b 100644 --- a/.env.example +++ b/.env.example @@ -86,3 +86,12 @@ CRE_LIBRARIAN_LINK_THRESHOLD=0.8 CRE_LIBRARIAN_BATCH_SIZE=32 CRE_LIBRARIAN_ECE_TARGET=0.10 CRE_LIBRARIAN_CONFORMAL_ALPHA=0.10 + +# Gap Analysis Opportunistic Upstream Fetch +# Enables local on-demand fetching of precomputed gap analysis pairs from prod. +# Set CRE_GA_NO_UPSTREAM=1 to disable and rely solely on local cache/backfill. +# CRE_GA_NO_UPSTREAM=false +# CRE_GA_UPSTREAM_API_URL=https://opencre.org/rest/v1 +# CRE_GA_UPSTREAM_TIMEOUT_SECONDS=30 +# CRE_GA_UPSTREAM_MAX_ATTEMPTS=4 +# CRE_GA_UPSTREAM_RETRY_BACKOFF_SECONDS=2 diff --git a/application/tests/ga_upstream_fetch_test.py b/application/tests/ga_upstream_fetch_test.py new file mode 100644 index 000000000..c5a24834f --- /dev/null +++ b/application/tests/ga_upstream_fetch_test.py @@ -0,0 +1,302 @@ +import json +import logging +import os +import unittest +from unittest.mock import MagicMock, call, patch + +import requests + +from application.utils import ga_upstream_fetch + + +class MockResponse: + def __init__( + self, + status_code: int, + json_data: dict = None, + headers: dict = None, + ): + self.status_code = status_code + self._json_data = json_data or {} + self.headers = headers or {} + + def json(self): + return self._json_data + + +class GaUpstreamFetchTest(unittest.TestCase): + def setUp(self): + self.db = MagicMock() + self.db.gap_analysis_exists.return_value = False + self.db.get_gap_analysis_result.return_value = None + + # Clean environment before each test + for var in [ + "CRE_GA_NO_UPSTREAM", + "CRE_GA_UPSTREAM_API_URL", + "CRE_GA_UPSTREAM_TIMEOUT_SECONDS", + "CRE_GA_UPSTREAM_MAX_ATTEMPTS", + "CRE_GA_UPSTREAM_RETRY_BACKOFF_SECONDS", + ]: + if var in os.environ: + del os.environ[var] + + def test_env_defaults(self): + """Unset env -> timeout 30, max_attempts 4.""" + session = MagicMock() + session.get.return_value = MockResponse( + 200, {"result": {"some": "data", "links": [1]}} + ) + + ga_upstream_fetch.fetch_ga_upstream_json( + ["Standard A", "Standard B"], session=session + ) + + session.get.assert_called_once() + _, kwargs = session.get.call_args + self.assertEqual(kwargs.get("timeout"), 30.0) + + def test_fetch_success_upserts_material_row(self): + """Mock prod 200 -> add_gap_analysis_result called once with material JSON.""" + session = MagicMock() + material_result = {"header": "test", "links": [{"id": 1}]} + session.get.return_value = MockResponse( + 200, {"result": material_result} + ) + + with patch( + "application.utils.gap_analysis.primary_gap_analysis_payload_is_material", + return_value=True, + ): + status_code, body = ga_upstream_fetch.resolve_gap_analysis_pair( + ["Standard A", "Standard B"], self.db, session=session + ) + + self.assertEqual(status_code, 200) + self.assertEqual(body, {"result": material_result}) + self.db.add_gap_analysis_result.assert_called_once() + call_kwargs = self.db.add_gap_analysis_result.call_args[1] + self.assertIn("Standard A >> Standard B", call_kwargs["cache_key"]) + stored_obj = json.loads(call_kwargs["ga_object"]) + self.assertEqual(stored_obj["result"], material_result) + + def test_fetch_404_no_upsert(self): + """Mock prod 404 -> no DB write; error dict returned.""" + session = MagicMock() + session.get.return_value = MockResponse(404) + + status_code, body = ga_upstream_fetch.resolve_gap_analysis_pair( + ["Standard A", "Standard B"], self.db, session=session + ) + + self.assertEqual(status_code, 404) + self.assertEqual(body["error_code"], "ga_cache_miss") + self.assertFalse(body["try_again"]) + self.db.add_gap_analysis_result.assert_not_called() + + def test_fetch_404_no_retry(self): + """Mock prod 404 -> HTTP call count = 1.""" + session = MagicMock() + session.get.return_value = MockResponse(404) + + _, status, _ = ga_upstream_fetch.fetch_ga_upstream_json( + ["Standard A", "Standard B"], session=session + ) + + self.assertEqual(status, 404) + self.assertEqual(session.get.call_count, 1) + + def test_no_upstream_flag_skips_http(self): + """CRE_GA_NO_UPSTREAM=1 -> zero HTTP calls.""" + os.environ["CRE_GA_NO_UPSTREAM"] = "1" + session = MagicMock() + + status_code, body = ga_upstream_fetch.resolve_gap_analysis_pair( + ["Standard A", "Standard B"], self.db, session=session + ) + + self.assertEqual(status_code, 404) + self.assertEqual(body["error_code"], "ga_cache_miss") + self.assertEqual(session.get.call_count, 0) + + def test_v1_no_refetch(self): + """Pre-seeded cache -> resolve -> HTTP mock not called.""" + self.db.gap_analysis_exists.return_value = True + cached_result = {"header": "already cached", "items": [1]} + self.db.get_gap_analysis_result.return_value = json.dumps( + {"result": cached_result} + ) + + session = MagicMock() + with patch( + "application.utils.gap_analysis.primary_gap_analysis_payload_is_material", + return_value=True, + ): + status_code, body = ga_upstream_fetch.resolve_gap_analysis_pair( + ["Standard A", "Standard B"], self.db, session=session + ) + + self.assertEqual(status_code, 200) + self.assertEqual(body, {"result": cached_result}) + self.assertEqual(session.get.call_count, 0) + + @patch("time.sleep", return_value=None) + def test_prod_down_serves_local_cache(self, _mock_sleep): + """Mock ConnectionError after retries + local row exists -> 200 from cache.""" + session = MagicMock() + session.get.side_effect = requests.exceptions.ConnectionError( + "Connection refused" + ) + + cached_result = {"header": "stale cached", "items": [2]} + + def fake_gap_analysis_exists(key): + return True + + self.db.gap_analysis_exists.side_effect = fake_gap_analysis_exists + # Simulate initial check missing or non-material, but local row available as fallback + self.db.get_gap_analysis_result.return_value = json.dumps( + {"result": cached_result} + ) + + with patch( + "application.utils.gap_analysis.primary_gap_analysis_payload_is_material", + side_effect=[False, True], + ): + status_code, body = ga_upstream_fetch.resolve_gap_analysis_pair( + ["Standard A", "Standard B"], self.db, session=session + ) + + self.assertEqual(status_code, 200) + self.assertEqual(body, {"result": cached_result}) + + @patch("time.sleep", return_value=None) + def test_prod_down_try_again(self, _mock_sleep): + """Mock ConnectionError + no local row -> 503 and try_again: true.""" + session = MagicMock() + session.get.side_effect = requests.exceptions.ConnectionError( + "Connection refused" + ) + + status_code, body = ga_upstream_fetch.resolve_gap_analysis_pair( + ["Standard A", "Standard B"], self.db, session=session + ) + + self.assertEqual(status_code, 503) + self.assertEqual(body["error_code"], "ga_upstream_unavailable") + self.assertTrue(body["try_again"]) + + @patch("time.sleep", return_value=None) + def test_failed_fetch_preserves_cache(self, _mock_sleep): + """Bad prod response does not overwrite cache.""" + session = MagicMock() + session.get.return_value = MockResponse(500) + + status_code, _ = ga_upstream_fetch.resolve_gap_analysis_pair( + ["Standard A", "Standard B"], self.db, session=session + ) + + self.db.add_gap_analysis_result.assert_not_called() + + def test_concurrent_in_progress(self): + """When in-flight lock exists, returns 202 in_progress.""" + session = MagicMock() + cache_key = "Standard A >> Standard B" + ga_upstream_fetch._IN_FLIGHT_KEYS.add(cache_key) + try: + status_code, body = ga_upstream_fetch.resolve_gap_analysis_pair( + ["Standard A", "Standard B"], self.db, session=session + ) + self.assertEqual(status_code, 202) + self.assertEqual(body["status"], "in_progress") + self.assertEqual(body["cache_key"], cache_key) + self.assertEqual(session.get.call_count, 0) + finally: + ga_upstream_fetch._IN_FLIGHT_KEYS.discard(cache_key) + + def test_soft_cta_positive(self): + """Both standards local + GA-eligible + prod 404 -> soft_cta.show: true.""" + session = MagicMock() + session.get.return_value = MockResponse(404) + + self.db.get_node.return_value = MagicMock() + self.db.get_ga_standards.return_value = [ + "Standard A", + "Standard B", + "Standard C", + ] + + status_code, body = ga_upstream_fetch.resolve_gap_analysis_pair( + ["Standard A", "Standard B"], self.db, session=session + ) + + self.assertEqual(status_code, 404) + self.assertIn("soft_cta", body) + self.assertTrue(body["soft_cta"]["show"]) + + def test_soft_cta_negative_missing_standard(self): + """One standard missing locally -> soft_cta.show: false.""" + session = MagicMock() + session.get.return_value = MockResponse(404) + + # First standard exists, second does not + self.db.get_node.side_effect = [MagicMock(), None] + self.db.get_ga_standards.return_value = ["Standard A", "Standard B"] + + status_code, body = ga_upstream_fetch.resolve_gap_analysis_pair( + ["Standard A", "Standard B"], self.db, session=session + ) + + self.assertEqual(status_code, 404) + self.assertNotIn("soft_cta", body) + + def test_soft_cta_negative_non_participant(self): + """Standard not in ga_standards -> soft_cta.show: false.""" + session = MagicMock() + session.get.return_value = MockResponse(404) + + self.db.get_node.return_value = MagicMock() + self.db.get_ga_standards.return_value = ["Standard C", "Standard D"] + + status_code, body = ga_upstream_fetch.resolve_gap_analysis_pair( + ["Standard A", "Standard B"], self.db, session=session + ) + + self.assertEqual(status_code, 404) + self.assertNotIn("soft_cta", body) + + def test_success_logs_upstream_pull(self): + """Successful pull logs INFO line.""" + session = MagicMock() + session.get.return_value = MockResponse( + 200, {"result": {"header": "foo", "links": [1]}} + ) + + with self.assertLogs("application.utils.ga_upstream_fetch", level="INFO") as cm: + with patch( + "application.utils.gap_analysis.primary_gap_analysis_payload_is_material", + return_value=True, + ): + ga_upstream_fetch.resolve_gap_analysis_pair( + ["Standard A", "Standard B"], self.db, session=session + ) + + self.assertTrue(any("GA upstream fetch: pulled pair" in line for line in cm.output)) + + def test_read_only_get(self): + """Ensure client only calls GET, never POST/PUT/DELETE.""" + session = MagicMock() + session.get.return_value = MockResponse(200, {"result": {}}) + + ga_upstream_fetch.fetch_ga_upstream_json( + ["Standard A", "Standard B"], session=session + ) + + session.get.assert_called_once() + session.post.assert_not_called() + session.put.assert_not_called() + session.delete.assert_not_called() + + +if __name__ == "__main__": + unittest.main() diff --git a/application/tests/web_main_test.py b/application/tests/web_main_test.py index 13b30f921..9d9999fc2 100644 --- a/application/tests/web_main_test.py +++ b/application/tests/web_main_test.py @@ -818,15 +818,12 @@ def test_gap_analysis_returns_existing_inflight_job( enqueue_call_mock.assert_not_called() @patch.object(db, "Node_collection") - @patch.object(rq.Queue, "enqueue_call") - @patch.object(app_redis, "connect") - def test_gap_analysis_create_job_id( - self, redis_conn_mock, enqueue_call_mock, db_mock + @patch.object(web_main.ga_upstream_fetch, "resolve_gap_analysis_pair") + def test_map_analysis_cache_miss_calls_upstream_resolver( + self, resolve_mock, db_mock ) -> None: - expected = {"job_id": "ABC"} - redis_conn_mock.return_value.get.return_value = None - enqueue_call_mock.return_value = Mock(id="ABC") - db_mock.return_value.get_gap_analysis_result.return_value = None + expected = {"result": {"standard_a": {}, "standard_b": {}}} + resolve_mock.return_value = (200, expected) db_mock.return_value.gap_analysis_exists.return_value = False with self.app.test_client() as client: response = client.get( @@ -835,76 +832,67 @@ def test_gap_analysis_create_job_id( ) self.assertEqual(200, response.status_code) self.assertEqual(expected, json.loads(response.data)) - self.assertTrue(enqueue_call_mock.called) - _, kwargs = enqueue_call_mock.call_args - self.assertEqual("aaa->bbb", kwargs["description"]) - self.assertEqual( - "application.cmd.cre_main.run_gap_pair_job", kwargs["func"] - ) - self.assertEqual("aaa", kwargs["kwargs"]["importing_name"]) - self.assertEqual("bbb", kwargs["kwargs"]["peer_name"]) - self.assertEqual(GAP_ANALYSIS_TIMEOUT, kwargs["timeout"]) + resolve_mock.assert_called_once() @patch.object(db, "Node_collection") - @patch.object(rq.Queue, "enqueue_call") - @patch.object(app_redis, "connect") - def test_map_analysis_non_material_sql_cache_triggers_job( - self, redis_conn_mock, enqueue_call_mock, db_mock + @patch.object(web_main.ga_upstream_fetch, "resolve_gap_analysis_pair") + def test_map_analysis_cache_miss_in_progress( + self, resolve_mock, db_mock ) -> None: - """SQL rows with empty ``result`` must not short-circuit map_analysis.""" - redis_conn_mock.return_value.get.return_value = None - enqueue_call_mock.return_value = Mock(id="ABC") + expected = { + "status": "in_progress", + "cache_key": "aaa >> bbb", + "poll_after_seconds": 2, + } + resolve_mock.return_value = (202, expected) db_mock.return_value.gap_analysis_exists.return_value = False - db_mock.return_value.get_gap_analysis_result.return_value = '{"result": {}}' with self.app.test_client() as client: response = client.get( "/rest/v1/map_analysis?standard=aaa&standard=bbb", headers={"Content-Type": "application/json"}, ) - self.assertEqual(200, response.status_code) - self.assertEqual({"job_id": "ABC"}, json.loads(response.data)) - enqueue_call_mock.assert_called_once() + self.assertEqual(202, response.status_code) + self.assertEqual(expected, json.loads(response.data)) @patch.object(db, "Node_collection") - @patch.object(db, "gap_analysis") - @patch.object(app_redis, "connect") - def test_gap_analysis_fallback_without_redis( - self, redis_conn_mock, db_gap_analysis_mock, db_mock + @patch.object(web_main.ga_upstream_fetch, "resolve_gap_analysis_pair") + def test_map_analysis_cache_miss_upstream_unavailable_returns_503( + self, resolve_mock, db_mock ) -> None: - redis_conn_mock.side_effect = RuntimeError("redis down") - db_mock.return_value.get_gap_analysis_result.side_effect = [ - None, - json.dumps({"result": {"k": {}}}), - ] + expected = { + "error_code": "ga_upstream_unavailable", + "message": "Could not reach opencre.org", + "try_again": True, + } + resolve_mock.return_value = (503, expected) + db_mock.return_value.gap_analysis_exists.return_value = False with self.app.test_client() as client: response = client.get( "/rest/v1/map_analysis?standard=aaa&standard=bbb", headers={"Content-Type": "application/json"}, ) - self.assertEqual(200, response.status_code) - self.assertEqual({"result": {"k": {}}}, json.loads(response.data)) - db_gap_analysis_mock.assert_called_once() + self.assertEqual(503, response.status_code) + self.assertEqual(expected, json.loads(response.data)) @patch.object(db, "Node_collection") - @patch.object(db, "gap_analysis") - @patch.object(app_redis, "connect") - def test_gap_analysis_fallback_backend_failure_returns_503( - self, redis_conn_mock, db_gap_analysis_mock, db_mock + @patch.object(web_main.ga_upstream_fetch, "resolve_gap_analysis_pair") + def test_map_analysis_cache_miss_returns_404( + self, resolve_mock, db_mock ) -> None: - redis_conn_mock.side_effect = RuntimeError("redis down") - db_gap_analysis_mock.side_effect = RuntimeError("neo unavailable") - db_mock.return_value.get_gap_analysis_result.return_value = None + expected = { + "error_code": "ga_cache_miss", + "message": "Not found", + "try_again": False, + } + resolve_mock.return_value = (404, expected) db_mock.return_value.gap_analysis_exists.return_value = False with self.app.test_client() as client: - with patch.dict(os.environ, {}, clear=False): - os.environ.pop("DYNO", None) - os.environ.pop("HEROKU", None) - response = client.get( - "/rest/v1/map_analysis?standard=aaa&standard=bbb", - headers={"Content-Type": "application/json"}, - ) - self.assertEqual(503, response.status_code) - db_gap_analysis_mock.assert_called_once() + response = client.get( + "/rest/v1/map_analysis?standard=aaa&standard=bbb", + headers={"Content-Type": "application/json"}, + ) + self.assertEqual(404, response.status_code) + self.assertEqual(expected, json.loads(response.data)) @patch.dict(os.environ, {"HEROKU": "True"}, clear=False) @patch.object(db, "Node_collection") diff --git a/application/utils/ga_upstream_fetch.py b/application/utils/ga_upstream_fetch.py new file mode 100644 index 000000000..83f5ebcda --- /dev/null +++ b/application/utils/ga_upstream_fetch.py @@ -0,0 +1,297 @@ +from __future__ import annotations + +import json +import os +import threading +import time +import urllib.parse +from typing import Any, Dict, List, Optional, Tuple + +import requests +from cre_logging import get_logger + +from application.database import db +from application.defs import cre_defs as defs +from application.utils import gap_analysis + +logger = get_logger(__name__) + +DEFAULT_UPSTREAM_API_URL = "https://opencre.org/rest/v1" +DEFAULT_TIMEOUT_SECONDS = 30.0 +DEFAULT_MAX_ATTEMPTS = 4 +DEFAULT_RETRY_BACKOFF_SECONDS = 2.0 +BACKFILL_DOCS_URL = ( + "https://github.com/OWASP/OpenCRE/blob/main/docs/backfill_gap_analysis.md" +) +GITHUB_NEW_ISSUE_URL = "https://github.com/OWASP/OpenCRE/issues/new" + +_IN_PROCESS_LOCK = threading.Lock() +_IN_FLIGHT_KEYS: set[str] = set() + + +def is_ga_upstream_disabled() -> bool: + """Return True if CRE_GA_NO_UPSTREAM is set to 1, true, or yes (case-insensitive).""" + val = os.environ.get("CRE_GA_NO_UPSTREAM", "").strip().lower() + return val in ("1", "true", "yes") + + +def _acquire_inflight_lock(cache_key: str, ttl: int = 150) -> bool: + """Acquire an in-flight lock for cache_key. Return True if acquired, False if already in-flight.""" + try: + from application.database import redis as cre_redis + + conn = cre_redis.connect() + lock_key = f"ga:upstream:inflight:{cache_key}" + res = conn.set(lock_key, "1", ex=ttl, nx=True) + if res: + return True + return False + except Exception: + with _IN_PROCESS_LOCK: + if cache_key in _IN_FLIGHT_KEYS: + return False + _IN_FLIGHT_KEYS.add(cache_key) + return True + + +def _release_inflight_lock(cache_key: str) -> None: + """Release the in-flight lock for cache_key.""" + try: + from application.database import redis as cre_redis + + conn = cre_redis.connect() + lock_key = f"ga:upstream:inflight:{cache_key}" + conn.delete(lock_key) + except Exception: + pass + with _IN_PROCESS_LOCK: + _IN_FLIGHT_KEYS.discard(cache_key) + + +def fetch_ga_upstream_json( + standards: List[str], + base_url: Optional[str] = None, + timeout: Optional[float] = None, + max_attempts: Optional[int] = None, + backoff_seconds: Optional[float] = None, + session: Optional[Any] = None, +) -> Tuple[Optional[Dict[str, Any]], Optional[int], Optional[Exception]]: + """ + Fetch GA pair from upstream prod with retries matching D12 / RFC #951 pattern. + Read-only GET requests, no auth headers. + Returns: (payload_dict, status_code, exception) + """ + if len(standards) < 2: + raise ValueError("Must provide at least two standards") + + base = ( + base_url + or os.environ.get("CRE_GA_UPSTREAM_API_URL", DEFAULT_UPSTREAM_API_URL) + ).rstrip("/") + req_timeout = timeout or float( + os.environ.get( + "CRE_GA_UPSTREAM_TIMEOUT_SECONDS", str(DEFAULT_TIMEOUT_SECONDS) + ) + ) + attempts = max_attempts or int( + os.environ.get( + "CRE_GA_UPSTREAM_MAX_ATTEMPTS", str(DEFAULT_MAX_ATTEMPTS) + ) + ) + backoff = backoff_seconds or float( + os.environ.get( + "CRE_GA_UPSTREAM_RETRY_BACKOFF_SECONDS", + str(DEFAULT_RETRY_BACKOFF_SECONDS), + ) + ) + + params = [("standard", standards[0]), ("standard", standards[1])] + query_string = urllib.parse.urlencode(params) + url = f"{base}/map_analysis?{query_string}" + + http = session or requests + last_exc: Optional[Exception] = None + last_status: Optional[int] = None + + for attempt in range(1, attempts + 1): + resp = None + try: + resp = http.get(url, timeout=req_timeout) + last_status = resp.status_code + if last_status == 200: + return resp.json(), 200, None + if last_status == 404: + # D12: Do not retry 404 + return None, 404, None + if last_status < 500 and last_status != 429: + # Other 4xx errors are terminal (do not retry) + return ( + None, + last_status, + RuntimeError(f"Upstream returned HTTP {last_status}"), + ) + # 429 or 5xx -> retry + last_exc = RuntimeError(f"Upstream returned HTTP {last_status}") + except requests.exceptions.RequestException as exc: + last_exc = exc + + if attempt < attempts: + sleep_time = backoff * attempt + if resp is not None and resp.status_code == 429: + hdr = resp.headers.get("Retry-After") + try: + if hdr: + sleep_time = float(hdr) + except (TypeError, ValueError): + pass + logger.warning( + "GA upstream fetch failed for %s on attempt %s/%s, retrying", + url, + attempt, + attempts, + ) + time.sleep(sleep_time) + + return None, last_status, last_exc + + +def _evaluate_soft_cta(standards: List[str], database: Any) -> Dict[str, Any]: + """ + Check D7 soft CTA conditions: + 1) Both standards exist locally in CRE graph + 2) Both are valid GA participants + 3) Upstream map_analysis exhausted with 404 + """ + try: + std_a, std_b = standards[0], standards[1] + node_a = database.get_node(name=std_a, ntype=defs.Standard.__name__) + node_b = database.get_node(name=std_b, ntype=defs.Standard.__name__) + if not node_a or not node_b: + return {"show": False} + + ga_stds: List[str] = [] + if hasattr(database, "get_ga_standards"): + ga_stds = database.get_ga_standards() + elif hasattr(database, "get_standards"): + ga_stds = [s.name for s in database.get_standards()] + + if ga_stds and (std_a not in ga_stds or std_b not in ga_stds): + return {"show": False} + + return { + "show": True, + "message": "Both standards appear valid for gap analysis but this pair is missing from opencre.org. Consider filing an issue.", + "issue_url": GITHUB_NEW_ISSUE_URL, + } + except Exception as ex: + logger.debug("Error checking soft CTA: %s", ex) + return {"show": False} + + +def resolve_gap_analysis_pair( + standards: List[str], + database: db.Node_collection, + session: Optional[Any] = None, +) -> Tuple[int, Dict[str, Any]]: + """ + Single shared backend resolver for local gap analysis on-demand fetch (D1, D2, D3, D4, D5, D7): + 1. Local cache hit: return cached row if material (V1 no re-fetch). + 2. CRE_GA_NO_UPSTREAM: return 404 + backfill instructions (no prod HTTP). + 3. In-flight lock: return 202 in_progress if another request is already fetching this pair. + 4. Fetch from upstream prod (opencre.org): + - 200: upsert material payload locally into gap_analysis_results, return 200. + - 404: check D7 soft CTA, return 404 + backfill docs. + - Exhausted retries / unreachable: return local cached row if present; else 503 try_again. + """ + if len(standards) < 2: + return 400, {"message": "Please provide two standards"} + standards = standards[:2] + cache_key = gap_analysis.make_resources_key(standards) + + # 1. Local cache check + if database.gap_analysis_exists(cache_key): + cached = database.get_gap_analysis_result(cache_key=cache_key) + if cached: + try: + parsed = json.loads(cached) + res = parsed.get("result", parsed) + if gap_analysis.primary_gap_analysis_payload_is_material(res): + return 200, {"result": res} + except Exception: + pass + + # 2. Disabled check + if is_ga_upstream_disabled(): + return 404, { + "error_code": "ga_cache_miss", + "message": "This gap analysis pair is not cached locally or on opencre.org.", + "backfill_docs_url": BACKFILL_DOCS_URL, + "try_again": False, + } + + # 3. In-flight lock check + acquired = _acquire_inflight_lock(cache_key) + if not acquired: + return 202, { + "status": "in_progress", + "cache_key": cache_key, + "poll_after_seconds": 2, + } + + base_url = ( + os.environ.get("CRE_GA_UPSTREAM_API_URL", DEFAULT_UPSTREAM_API_URL) + ).rstrip("/") + try: + payload, status_code, exc = fetch_ga_upstream_json( + standards, + base_url=base_url, + session=session, + ) + + if status_code == 200 and payload: + result_data = payload.get("result", payload) + if gap_analysis.primary_gap_analysis_payload_is_material( + result_data + ): + database.add_gap_analysis_result( + cache_key=cache_key, + ga_object=json.dumps({"result": result_data}), + ) + logger.info( + "GA upstream fetch: pulled pair %s from %s", + cache_key, + base_url, + ) + return 200, {"result": result_data} + + if status_code == 404: + soft_cta = _evaluate_soft_cta(standards, database) + resp_body: Dict[str, Any] = { + "error_code": "ga_cache_miss", + "message": "This gap analysis pair is not cached locally or on opencre.org.", + "backfill_docs_url": BACKFILL_DOCS_URL, + "try_again": False, + } + if soft_cta.get("show"): + resp_body["soft_cta"] = soft_cta + return 404, resp_body + + # Unreachable or exhausted retries: serve local row if available + if database.gap_analysis_exists(cache_key): + cached = database.get_gap_analysis_result(cache_key=cache_key) + if cached: + try: + parsed = json.loads(cached) + if "result" in parsed: + return 200, {"result": parsed.get("result")} + except Exception: + pass + + return 503, { + "error_code": "ga_upstream_unavailable", + "message": "Could not reach opencre.org to fetch this gap analysis.", + "backfill_docs_url": BACKFILL_DOCS_URL, + "try_again": True, + } + finally: + _release_inflight_lock(cache_key) diff --git a/application/web/web_main.py b/application/web/web_main.py index 094f818e8..f9c7bbc38 100644 --- a/application/web/web_main.py +++ b/application/web/web_main.py @@ -30,7 +30,7 @@ ) from application.web import identity from application.utils import spreadsheet as sheet_utils -from application.utils import mdutils, redirectors, gap_analysis +from application.utils import mdutils, redirectors, gap_analysis, ga_upstream_fetch from application.web.openapi_registry import openapi_documented from enum import Enum from flask import json as flask_json @@ -408,63 +408,15 @@ def map_analysis() -> Any: ) abort(404, "No such Cache") - if os.environ.get("CRE_NO_CALCULATE_GAP_ANALYSIS"): - logger.info( - f"Gap analysis calculations are disabled by CRE_NO_CALCULATE_GAP_ANALYSIS; " - f"refusing to schedule new job for {standards_hash}" - ) - abort(404, "Gap analysis calculations are disabled") - - # ----- upstream: Redis / RQ path with synchronous fallback ----- - db_url = os.environ.get("CRE_CACHE_FILE") or os.environ.get("PROD_DATABASE_URL") - if not db_url: - db_url = str(getattr(getattr(database.session, "bind", None), "url", "")) - try: - conn = redis.connect() - ga_queue_name = os.environ.get("CRE_GA_QUEUE_NAME", "ga") - q = Queue(name=ga_queue_name, connection=conn) - inflight_key = f"ga:inflight:{cache_key}" - inflight_job_id_raw = conn.get(inflight_key) - inflight_job_id = ( - inflight_job_id_raw.decode("utf-8") - if isinstance(inflight_job_id_raw, bytes) - else str(inflight_job_id_raw) if inflight_job_id_raw else "" - ) - if inflight_job_id: - try: - inflight_job = job.Job.fetch(id=inflight_job_id, connection=conn) - if inflight_job.get_status() in ( - job.JobStatus.QUEUED, - job.JobStatus.STARTED, - ): - return jsonify({"job_id": inflight_job_id}) - except exceptions.NoSuchJobError: - conn.delete(inflight_key) - - j = q.enqueue_call( - description=f"{standards[0]}->{standards[1]}", - func="application.cmd.cre_main.run_gap_pair_job", - kwargs={ - "importing_name": standards[0], - "peer_name": standards[1], - "db_connection_str": db_url, - }, - timeout=gap_analysis.GAP_ANALYSIS_TIMEOUT, - ) - conn.set(inflight_key, str(j.id)) - conn.expire(inflight_key, _ga_timeout_seconds()) - return jsonify({"job_id": str(j.id)}) - except Exception as exc: - logger.warning( - "Redis/RQ unavailable in map_analysis for %s; using synchronous fallback: %s", - cache_key, - exc, - ) - try: - return jsonify(_compute_ga_without_redis(database, standards)) - except Exception as fallback_exc: - logger.exception("Synchronous GA fallback failed for %s", cache_key) - abort(503, f"Gap analysis unavailable: {fallback_exc}") + # Local / non-Heroku opportunistic upstream fetch (RFC #972 / #1026) + status_code, upstream_resp = ga_upstream_fetch.resolve_gap_analysis_pair( + standards, database + ) + if status_code == 200: + return jsonify(upstream_resp) + if status_code == 202: + return jsonify(upstream_resp), 202 + return jsonify(upstream_resp), status_code @openapi_documented("map_analysis_weak_links")