diff --git a/src/background_data_service.py b/src/background_data_service.py index 355c93c4..e6891d60 100644 --- a/src/background_data_service.py +++ b/src/background_data_service.py @@ -18,7 +18,7 @@ import time import logging import threading import requests -from typing import Dict, Any, Optional, Callable +from typing import Dict, Any, Optional, Callable, List from dataclasses import dataclass, field from enum import Enum import queue @@ -50,6 +50,11 @@ class FetchRequest: max_retries: int = 3 priority: int = 1 # Higher number = higher priority callback: Optional[Callable] = None + # Callbacks from submitters that JOINED this fetch instead of starting a + # duplicate one. The primary `callback` above belongs to whoever created + # the request; these belong to everyone who asked for the same cache_key + # while it was still in flight. + extra_callbacks: List[Callable] = field(default_factory=list) created_at: float = field(default_factory=time.time) status: FetchStatus = FetchStatus.PENDING result: Optional[Any] = None @@ -90,6 +95,14 @@ class BackgroundDataService: # Thread management self.executor = ThreadPoolExecutor(max_workers=max_workers, thread_name_prefix="BackgroundData") + # cache_key -> request_id for fetches currently in flight. Submitting + # the same key twice used to start two identical fetches: request_id + # carries a millisecond timestamp, so every submit looked new, and + # active_requests is keyed by it rather than by what is being fetched. + # On a real board the season-schedule key is requested by both the + # Recent and the Upcoming manager, which miss the cache in the same + # millisecond and each download and parse the same payload. + self._inflight_by_cache_key: Dict[str, str] = {} self.active_requests: Dict[str, FetchRequest] = {} self.completed_requests: Dict[str, FetchResult] = {} self.request_queue = queue.PriorityQueue() @@ -218,7 +231,29 @@ class BackgroundDataService: ) with self._lock: + existing_id = self._inflight_by_cache_key.get(cache_key) + existing = self.active_requests.get(existing_id) if existing_id else None + if existing_id and existing is None: + # Stranded index entry: the request it names is gone. Drop it and + # fetch normally. Looking the request up rather than trusting the + # id is what stops a stale entry wedging a key forever. + del self._inflight_by_cache_key[cache_key] + if existing is not None: + # Someone is already fetching this key. Ride along rather than + # duplicating the download, the parse and the resident copy. + if callback: + existing.extra_callbacks.append(callback) + self.stats['deduplicated_requests'] = ( + self.stats.get('deduplicated_requests', 0) + 1 + ) + logger.info( + "Joined in-flight fetch %s for %s (cache_key=%s) instead of " + "starting a duplicate", existing_id, sport, cache_key + ) + return existing_id + self.active_requests[request_id] = request + self._inflight_by_cache_key[cache_key] = request_id self.stats['total_requests'] += 1 self.stats['cache_misses'] += 1 @@ -311,6 +346,16 @@ class BackgroundDataService: self.completed_requests[request.id] = result if request.id in self.active_requests: del self.active_requests[request.id] + # Stop accepting joiners and take the callback list in the same + # critical section. A submitter that arrives after this point + # finds no in-flight entry and either hits the cache (written + # above, before the result was built) or starts a fresh fetch -- + # what it must never do is join a fetch whose callbacks have + # already run and then never be called. + if self._inflight_by_cache_key.get(request.cache_key) == request.id: + del self._inflight_by_cache_key[request.cache_key] + callbacks = ([request.callback] if request.callback else []) + callbacks.extend(request.extra_callbacks) # Update statistics if result.success: @@ -327,10 +372,11 @@ class BackgroundDataService: # Periodic cleanup after storing result self._cleanup_completed_requests() - # Call callback if provided - if request.callback: + # Call every callback: the original submitter's and any that joined + # this fetch. One raising must not stop the others being delivered. + for cb in callbacks: try: - request.callback(result) + cb(result) except Exception as e: logger.error(f"Error in callback for request {request.id}: {e}") @@ -440,6 +486,11 @@ class BackgroundDataService: request = self.active_requests[request_id] request.status = FetchStatus.CANCELLED del self.active_requests[request_id] + # Cancelling is the other way a request leaves active_requests, + # so the in-flight index has to be released here too or the key + # stays pointed at a request that no longer exists. + if self._inflight_by_cache_key.get(request.cache_key) == request_id: + del self._inflight_by_cache_key[request.cache_key] logger.info(f"Cancelled request {request_id}") return True return False diff --git a/test/test_background_fetch_dedupe.py b/test/test_background_fetch_dedupe.py new file mode 100644 index 00000000..6438c6cf --- /dev/null +++ b/test/test_background_fetch_dedupe.py @@ -0,0 +1,212 @@ +"""A second request for a key already being fetched must join, not duplicate. + +request_id embeds a millisecond timestamp and active_requests is keyed by it, +so every submit looked new and nothing compared what was actually being +fetched. On a real board the season-schedule cache_key is requested by both +the Recent and the Upcoming manager: they miss the cache in the same +millisecond and each start a full download and parse of the same payload. +Measured on a running board, 138 background fetches in 24 hours arriving in +pairs at identical timestamps -- half of them redundant. + +The cost of a duplicate is a second download, a second JSON parse (the +expensive part on a Pi), and a second parsed copy resident at the same time. +Schedules on that board run from 256KB to 20MB. It also consumes a second of +the three executor slots with identical work, which is what makes two large +parses peak simultaneously. +""" + +import threading +import time +from unittest.mock import MagicMock, Mock, patch + +import pytest + +from src.background_data_service import BackgroundDataService + + +PAYLOAD = {"events": [{"id": f"g{i}"} for i in range(20)]} + + +@pytest.fixture +def cache(): + m = MagicMock() + m.get.return_value = None # always a miss: force the fetch path + m.set.return_value = None + return m + + +@pytest.fixture +def service(cache): + svc = BackgroundDataService(cache, max_workers=3, request_timeout=5) + yield svc + svc.shutdown(wait=False) + + +def _resp(): + r = Mock() + r.json.return_value = PAYLOAD + r.raise_for_status.return_value = None + return r + + +def _wait(service, req_id, timeout=5): + deadline = time.time() + timeout + while not service.is_request_complete(req_id) and time.time() < deadline: + time.sleep(0.02) + + +class _BlockingSession: + """Holds the first fetch open so a second can be submitted mid-flight.""" + + def __init__(self): + self.calls = 0 + self.release = threading.Event() + self.started = threading.Event() + + def get(self, *a, **k): + self.calls += 1 + self.started.set() + self.release.wait(timeout=5) + return _resp() + + +def test_a_second_submit_for_the_same_key_does_not_fetch_twice(service): + session = _BlockingSession() + with patch.object(service, "session", session): + first = service.submit_fetch_request( + sport="nba", year=2026, url="https://x/s", cache_key="nba_2026", + callback=lambda r: None, max_retries=0) + assert session.started.wait(timeout=5) + + second = service.submit_fetch_request( + sport="nba", year=2026, url="https://x/s", cache_key="nba_2026", + callback=lambda r: None, max_retries=0) + + assert second == first, "the joiner should share the in-flight request id" + session.release.set() + _wait(service, first) + + assert session.calls == 1, f"the payload was fetched {session.calls} times" + + +def test_the_joiner_still_gets_its_callback(service): + session = _BlockingSession() + seen = [] + with patch.object(service, "session", session): + first = service.submit_fetch_request( + sport="nba", year=2026, url="https://x/s", cache_key="k", + callback=lambda r: seen.append("first"), max_retries=0) + assert session.started.wait(timeout=5) + joined = service.submit_fetch_request( + sport="nba", year=2026, url="https://x/s", cache_key="k", + callback=lambda r: seen.append("second"), max_retries=0) + # Assert the coalescing happened, otherwise this passes trivially: + # two independent requests would each fire their own callback and the + # test would say nothing about the joined path. + assert joined == first + session.release.set() + _wait(service, first) + + deadline = time.time() + 5 + while len(seen) < 2 and time.time() < deadline: + time.sleep(0.02) + assert sorted(seen) == ["first", "second"], ( + f"both submitters must be called back, got {seen}") + + +def test_one_callback_raising_does_not_silence_the_other(service): + session = _BlockingSession() + seen = [] + + def boom(result): + raise RuntimeError("consumer blew up") + + with patch.object(service, "session", session): + first = service.submit_fetch_request( + sport="nba", year=2026, url="https://x/s", cache_key="k", + callback=boom, max_retries=0) + assert session.started.wait(timeout=5) + joined = service.submit_fetch_request( + sport="nba", year=2026, url="https://x/s", cache_key="k", + callback=lambda r: seen.append("survivor"), max_retries=0) + # Same reason: without coalescing these are separate requests and + # neither callback can affect the other. + assert joined == first + session.release.set() + _wait(service, first) + + deadline = time.time() + 5 + while not seen and time.time() < deadline: + time.sleep(0.02) + assert seen == ["survivor"] + + +def test_different_keys_are_not_coalesced(service): + session = _BlockingSession() + with patch.object(service, "session", session): + a = service.submit_fetch_request( + sport="nba", year=2026, url="https://x/a", cache_key="key_a", + callback=lambda r: None, max_retries=0) + assert session.started.wait(timeout=5) + b = service.submit_fetch_request( + sport="nhl", year=2026, url="https://x/b", cache_key="key_b", + callback=lambda r: None, max_retries=0) + assert a != b, "different cache keys must not share a request" + session.release.set() + _wait(service, a) + _wait(service, b) + assert session.calls == 2 + + +def test_a_later_submit_after_completion_fetches_again(service): + """Dedupe is for concurrent requests only, not a second cache layer.""" + with patch.object(service.session, "get", side_effect=[_resp(), _resp()]) as get: + first = service.submit_fetch_request( + sport="nba", year=2026, url="https://x/s", cache_key="k", + callback=lambda r: None, max_retries=0) + _wait(service, first) + second = service.submit_fetch_request( + sport="nba", year=2026, url="https://x/s", cache_key="k", + callback=lambda r: None, max_retries=0) + _wait(service, second) + assert first != second + assert get.call_count == 2 + + +def test_cancelling_releases_the_key(service): + """A cancelled request must not wedge its key against future fetches.""" + session = _BlockingSession() + with patch.object(service, "session", session): + first = service.submit_fetch_request( + sport="nba", year=2026, url="https://x/s", cache_key="k", + callback=lambda r: None, max_retries=0) + assert session.started.wait(timeout=5) + service.cancel_request(first) + assert "k" not in service._inflight_by_cache_key + session.release.set() + + +def test_a_stranded_index_entry_cannot_wedge_a_key(service): + """Defensive: the request is looked up, not trusted from the id alone.""" + service._inflight_by_cache_key["ghost"] = "no_such_request" + with patch.object(service.session, "get", return_value=_resp()): + req = service.submit_fetch_request( + sport="nba", year=2026, url="https://x/s", cache_key="ghost", + callback=lambda r: None, max_retries=0) + _wait(service, req) + assert service.get_result(req).success is True + + +def test_the_deduplicated_count_is_reported(service): + session = _BlockingSession() + with patch.object(service, "session", session): + first = service.submit_fetch_request( + sport="nba", year=2026, url="https://x/s", cache_key="k", + callback=lambda r: None, max_retries=0) + assert session.started.wait(timeout=5) + service.submit_fetch_request( + sport="nba", year=2026, url="https://x/s", cache_key="k", + max_retries=0) + session.release.set() + _wait(service, first) + assert service.get_statistics().get("deduplicated_requests") == 1