From 15efac019f5279511de0a9b37250310c910e4726 Mon Sep 17 00:00:00 2001 From: chreman Date: Thu, 16 Jul 2026 10:14:04 +0200 Subject: [PATCH] refactor functionality of content provider lookup caching --- server/workers/api/src/apis/base.py | 4 +- .../tests/unit/test_contentprovider_cache.py | 187 ++++++++++++++++++ server/workers/base/src/base.py | 15 +- server/workers/base/tests/unit/test_base.py | 12 ++ server/workers/common/common/utils.py | 170 +++++++++++++--- 5 files changed, 362 insertions(+), 26 deletions(-) create mode 100644 server/workers/api/tests/unit/test_contentprovider_cache.py diff --git a/server/workers/api/src/apis/base.py b/server/workers/api/src/apis/base.py index a45fcf81a..2c4fc2ec2 100644 --- a/server/workers/api/src/apis/base.py +++ b/server/workers/api/src/apis/base.py @@ -9,11 +9,11 @@ from .request_validators import SearchParamSchema from common.utils import get_key, redis_store, get_or_create_contentprovider_lookup -contentprovider_lookup = get_or_create_contentprovider_lookup() - base_ns = Namespace("base", description="BASE API operations") search_param_schema = SearchParamSchema() +contentprovider_lookup = get_or_create_contentprovider_lookup(logger=base_ns.logger) + base_querymodel = base_ns.model("SearchQuery", {"q": fields.String(example='feminicide', diff --git a/server/workers/api/tests/unit/test_contentprovider_cache.py b/server/workers/api/tests/unit/test_contentprovider_cache.py new file mode 100644 index 000000000..9f282b6f2 --- /dev/null +++ b/server/workers/api/tests/unit/test_contentprovider_cache.py @@ -0,0 +1,187 @@ +import json +import time +import threading +from contextlib import contextmanager + +import pytest +from redis.exceptions import LockError + +import common.utils as cu +from common.utils import ( + get_contentprovider_records, + get_or_create_contentprovider_lookup, + CONTENTPROVIDER_RECORDS_KEY, +) + +RECORDS = [ + {"name": "Université de Lausanne", "internal_name": "ftunivlausanne"}, + {"name": "Some Repo", "internal_name": "ftsomerepo"}, +] + + +# --- Minimal in-memory Redis double supporting get / set(ex) / lock ----------- + +class FakeRedis: + def __init__(self, fail_lock=False): + self.store = {} + self.last_ex = None + self._fail_lock = fail_lock + self._lock = threading.Lock() + + def get(self, key): + return self.store.get(key) + + def set(self, key, value, ex=None): + self.store[key] = value + self.last_ex = ex + + @contextmanager + def lock(self, name, timeout=None, blocking_timeout=None): + if self._fail_lock: + raise LockError("could not acquire lock") + acquired = self._lock.acquire( + timeout=blocking_timeout if blocking_timeout is not None else -1 + ) + if not acquired: + raise LockError("lock acquisition timed out") + try: + yield + finally: + self._lock.release() + + +# --- get_contentprovider_records --------------------------------------------- + +def test_cache_hit_skips_producer(): + r = FakeRedis() + r.set(CONTENTPROVIDER_RECORDS_KEY, json.dumps(RECORDS)) + calls = [] + + out = get_contentprovider_records(r, lambda: calls.append(1) or []) + + assert out == RECORDS + assert calls == [] # producer never called on a warm cache + + +def test_cache_miss_produces_and_caches_with_ttl(): + r = FakeRedis() + calls = [] + + def produce(): + calls.append(1) + return RECORDS + + out = get_contentprovider_records(r, produce, ttl=1234) + + assert out == RECORDS + assert calls == [1] + assert json.loads(r.store[CONTENTPROVIDER_RECORDS_KEY]) == RECORDS + assert r.last_ex == 1234 + + +def test_single_producer_under_contention(): + r = FakeRedis() + counter = {"n": 0} + counter_lock = threading.Lock() + + def produce(): + with counter_lock: + counter["n"] += 1 + time.sleep(0.2) # hold the lock long enough for others to contend + return RECORDS + + results = [] + + def worker(): + results.append(get_contentprovider_records(r, produce)) + + threads = [threading.Thread(target=worker) for _ in range(5)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert counter["n"] == 1 # exactly one fetch despite 5 concurrent callers + assert all(res == RECORDS for res in results) + + +def test_lock_contention_waits_for_published_cache(): + r = FakeRedis(fail_lock=True) # this caller can never acquire the lock + + def publish_later(): + time.sleep(0.1) + r.set(CONTENTPROVIDER_RECORDS_KEY, json.dumps(RECORDS)) + + t = threading.Thread(target=publish_later) + t.start() + + def produce_should_not_run(): + raise AssertionError("producer must not run when the lock is held elsewhere") + + out = get_contentprovider_records(r, produce_should_not_run, poll_timeout=5) + t.join() + + assert out == RECORDS + + +def test_producer_error_falls_back_to_bundled(): + r = FakeRedis() + + def produce(): + raise RuntimeError("boom") + + out = get_contentprovider_records(r, produce) + + assert isinstance(out, list) and len(out) > 0 + assert {"name", "internal_name"} <= set(out[0].keys()) + + +def test_corrupt_cache_value_is_reproduced(): + r = FakeRedis() + r.set(CONTENTPROVIDER_RECORDS_KEY, "not-json{") + + out = get_contentprovider_records(r, lambda: RECORDS) + + assert out == RECORDS + + +def test_cache_hit_logs_debug_trace(caplog): + r = FakeRedis() + r.set(CONTENTPROVIDER_RECORDS_KEY, json.dumps(RECORDS)) + + with caplog.at_level("DEBUG"): + get_contentprovider_records(r, lambda: []) + + assert any( + "contentprovider-cache: served" in rec.message for rec in caplog.records + ) + + +def test_fallback_logs_warning(caplog): + r = FakeRedis() + + def produce(): + raise RuntimeError("boom") + + with caplog.at_level("WARNING"): + get_contentprovider_records(r, produce) + + assert any( + rec.levelname == "WARNING" and "bundled fallback" in rec.message + for rec in caplog.records + ) + + +# --- get_or_create_contentprovider_lookup (API forward map) ------------------ + +def test_get_or_create_builds_forward_lookup(monkeypatch): + monkeypatch.setattr( + cu, "get_contentprovider_records", lambda store, fn, **kw: RECORDS + ) + + lookup = get_or_create_contentprovider_lookup() + + assert lookup == { + "ftunivlausanne": "Université de Lausanne", + "ftsomerepo": "Some Repo", + } diff --git a/server/workers/base/src/base.py b/server/workers/base/src/base.py index a9b23c7f4..b9784b308 100644 --- a/server/workers/base/src/base.py +++ b/server/workers/base/src/base.py @@ -27,6 +27,7 @@ import dateparser import sys from common.rate_limiter import RateLimiter +from common.utils import get_contentprovider_records class BaseClient(RWrapper): @@ -35,8 +36,11 @@ def __init__(self, *args): self.rate_limiter = RateLimiter(self.redis_store, "base-ratelimit", 1.5) try: - result = self.get_contentproviders() - df = pd.DataFrame(json.loads(result["contentproviders"])) + records = get_contentprovider_records( + self.redis_store, self._fetch_contentprovider_records, + logger=self.logger + ) + df = pd.DataFrame(records) df.set_index("name", inplace=True) cp_dict = df.internal_name.to_dict() self.content_providers = cp_dict @@ -156,6 +160,13 @@ def enrich_metadata(self, metadata): metadata = pd.concat([metadata, enrichment], axis=1) return metadata + def _fetch_contentprovider_records(self): + """Run the R fetch and return the parsed list of content provider records.""" + result = self.get_contentproviders() + if result.get("status") == "error": + raise RuntimeError("contentproviders fetch returned an error") + return json.loads(result["contentproviders"]) + def get_contentproviders(self): runner = os.path.abspath(os.path.join(self.wd, "run_base_contentproviders.R")) cmd = [self.command, runner, self.wd] diff --git a/server/workers/base/tests/unit/test_base.py b/server/workers/base/tests/unit/test_base.py index 675572a70..13543e49b 100644 --- a/server/workers/base/tests/unit/test_base.py +++ b/server/workers/base/tests/unit/test_base.py @@ -178,6 +178,18 @@ def dummy_popen_cp(cmd, stdin, stdout, stderr, encoding): assert cp_list[0]["name"] == "cp1" assert cp_list[0]["internal_name"] == "Provider1" +def test_fetch_contentprovider_records_parses(client_base): + # get_contentproviders is stubbed by the fixture to a fixed payload. + records = client_base._fetch_contentprovider_records() + assert records == [{"name": "cp1", "internal_name": "Provider1"}] + + +def test_fetch_contentprovider_records_raises_on_error(client_base): + client_base.get_contentproviders = lambda: {"status": "error"} + with pytest.raises(RuntimeError): + client_base._fetch_contentprovider_records() + + # --- Tests for parser functions --- def test_filter_duplicates(): diff --git a/server/workers/common/common/utils.py b/server/workers/common/common/utils.py index 70e0757e6..08f7d0da1 100644 --- a/server/workers/common/common/utils.py +++ b/server/workers/common/common/utils.py @@ -5,8 +5,10 @@ import uuid import redis import pathlib +import logging import numpy as np import pandas as pd +from redis.exceptions import LockError from datetime import timedelta from dateutil.parser import parse from typing import Dict, List, Union @@ -81,30 +83,154 @@ def detect_error(service, error, params): return code, reason -def get_or_create_contentprovider_lookup(): +# Shared, cross-worker cache for the BASE content provider list. The list is +# expensive to fetch (external BASE call via an R script), so instead of every +# gunicorn worker / BASE worker fetching its own copy on startup, the first +# process to see a cold cache fetches it under a Redis lock and publishes it for +# all others to reuse. Same lock idiom as common.rate_limiter.RateLimiter. +CONTENTPROVIDER_RECORDS_KEY = "base:contentproviders:records" +CONTENTPROVIDER_LOCK_KEY = "lock:base:contentproviders" +CONTENTPROVIDER_CACHE_TTL = 86400 # seconds (24h) + +# All cache-lifecycle lines share this prefix so they can be grepped/counted in +# aggregated logs (e.g. `grep 'contentprovider-cache:'`). +_CP_LOG_PREFIX = "contentprovider-cache:" +_cp_logger = logging.getLogger(__name__) + + +def _load_bundled_contentprovider_records(): + """Read the content provider list bundled with the package (last-resort fallback).""" + df = pd.read_json( + pathlib.Path(__file__).parent.absolute() / "contentproviders.json" + ) + return df.to_dict(orient="records") + + +def _read_cached_contentprovider_records(redis_store): + """Return the cached records list, or None if absent/unreadable.""" + cached = redis_store.get(CONTENTPROVIDER_RECORDS_KEY) + if cached is None: + return None try: - k = str(uuid.uuid4()) - d = {"id": k, "params": {},"endpoint": "contentproviders"} - redis_store.rpush("base", json.dumps(d)) - result = get_key(redis_store, k, 100) - if result.get("status") == "error": - df = pd.read_json("contentproviders.json") - df.set_index("internal_name", inplace=True) - cp_dict = df.name.to_dict() - return cp_dict - else: - df = pd.DataFrame(json.loads(result["contentproviders"])) - df.set_index("internal_name", inplace=True) - cp_dict = df.name.to_dict() - return cp_dict - except Exception as e: - df = pd.read_json( - pathlib.Path(__file__).parent.absolute() / - "contentproviders.json" + return json.loads(cached) + except (ValueError, TypeError): + return None + + +def _poll_cached_contentprovider_records(redis_store, timeout, wait_s=1): + """Wait for another process to publish the cache, up to `timeout` seconds.""" + tries = 0 + max_tries = timeout / wait_s + while tries <= max_tries: + records = _read_cached_contentprovider_records(redis_store) + if records is not None: + return records + time.sleep(wait_s) + tries += 1 + return None + + +def get_contentprovider_records( + redis_store, + produce_fn, + ttl=CONTENTPROVIDER_CACHE_TTL, + blocking_timeout=120, + lock_timeout=180, + poll_timeout=120, + logger=None, +): + """ + Return the BASE content provider records as a shared, Redis-cached resource. + + The list is fetched at most once per `ttl` window across all workers: the + first caller to see a cold cache acquires a lock, calls `produce_fn` to fetch + the records, and stores them in Redis. Concurrent callers either read the + freshly written key or fall back to the bundled list, so a simultaneous + restart of many workers triggers a single fetch instead of one per process. + + :param redis_store: Redis client shared across workers. + :param produce_fn: zero-arg callable returning a list of + {"name", "internal_name"} records (the expensive fetch). + :param ttl: cache lifetime in seconds. + :param logger: logger for the cache-lifecycle trace (defaults to the module + logger). Pass the caller's logger to surface the lines in its stream. + :return: list of content provider records. + """ + log = logger or _cp_logger + records = _read_cached_contentprovider_records(redis_store) + if records is not None: + log.debug("%s served %d records from shared cache", _CP_LOG_PREFIX, len(records)) + return records + try: + with redis_store.lock( + CONTENTPROVIDER_LOCK_KEY, + timeout=lock_timeout, + blocking_timeout=blocking_timeout, + ): + # Double-checked: another process may have populated the cache while + # we were waiting for the lock. + records = _read_cached_contentprovider_records(redis_store) + if records is not None: + log.debug("%s served %d records from shared cache (post-lock)", + _CP_LOG_PREFIX, len(records)) + return records + log.debug("%s cold cache, fetching under lock", _CP_LOG_PREFIX) + records = produce_fn() + redis_store.set( + CONTENTPROVIDER_RECORDS_KEY, json.dumps(records), ex=ttl + ) + log.debug("%s fetched and published %d records (ttl=%ss)", + _CP_LOG_PREFIX, len(records), ttl) + return records + except LockError: + # Another process holds the lock and is producing; wait for it to publish + # the cache before giving up. + records = _poll_cached_contentprovider_records(redis_store, poll_timeout) + if records is not None: + log.debug("%s served %d records published by another process", + _CP_LOG_PREFIX, len(records)) + return records + except Exception: + # produce_fn failed or Redis is unavailable; fall through to bundled. + log.warning("%s fetch failed, using bundled fallback", _CP_LOG_PREFIX, + exc_info=True) + records = _load_bundled_contentprovider_records() + return records + log.warning("%s no cache published within timeout, using bundled fallback", + _CP_LOG_PREFIX) + return _load_bundled_contentprovider_records() + + +def _fetch_contentprovider_records_via_queue(): + """Produce records by asking a BASE worker over the Redis queue (API side).""" + k = str(uuid.uuid4()) + d = {"id": k, "params": {}, "endpoint": "contentproviders"} + redis_store.rpush("base", json.dumps(d)) + result = get_key(redis_store, k, 100) + if result.get("status") == "error": + raise RuntimeError("contentproviders queue request returned an error") + return json.loads(result["contentproviders"]) + + +def get_or_create_contentprovider_lookup(logger=None): + """ + Build the {internal_name -> human readable name} lookup used by the API to + resolve the "data provider" label, backed by the shared Redis cache so all + gunicorn workers reuse a single fetch. + + :param logger: logger for the cache-lifecycle trace (defaults to the module + logger). The API passes its namespace logger so the state of the + request/retrieval shows up alongside the other API log lines. + """ + try: + records = get_contentprovider_records( + redis_store, _fetch_contentprovider_records_via_queue, logger=logger ) - df.set_index("internal_name", inplace=True) - cp_dict = df.name.to_dict() - return cp_dict + except Exception: + records = _load_bundled_contentprovider_records() + df = pd.DataFrame(records) + df.set_index("internal_name", inplace=True) + return df.name.to_dict() def get_nested_value(data, keys, default=None): """