fix(memory): join an in-flight fetch instead of starting a duplicate

submit_fetch_request() had no notion of "already fetching this". request_id
embeds a millisecond timestamp, so every submit looked new, and
active_requests is keyed by that id rather than by what is being fetched.
Two submits for the same cache_key therefore started two identical fetches.

It is not a rare race. _fetch_data in the sports managers branches: the Live
manager fetches only today's games, but Recent and Upcoming both pull the
full season schedule under the SAME cache_key. On a cache miss both miss,
both submit, and nothing stops the second. On a running 512x64 board:

    138 background fetches in 24 hours, arriving in pairs at identical
    millisecond timestamps, roughly hourly:

      2  2026-08-25 11:47:26.612
      2  2026-08-25 10:46:28.064
      2  2026-08-25 08:01:55.962

Half of them redundant. Each duplicate costs 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 256KB to 20MB, 106MB
across all sports. Because the pairs land in the same millisecond they also
occupy two of the three executor slots with identical work, which is what
makes two large parses peak simultaneously.

A submit for a cache_key already in flight now joins that request: its
callback is added to the existing one and the existing request_id is
returned, so get_result() works for both. Different keys are untouched, and
dedupe applies only while a fetch is in flight -- a submit after completion
fetches again, because this is not a second cache layer.

Three details:

  - The in-flight entry is dropped and the callback list snapshotted in the
    SAME critical section as filing the result. Otherwise a submitter could
    join a fetch whose callbacks had already run and never be called back.

  - Cancellation is the other way a request leaves active_requests, so it
    releases the key too. And the join path looks the request up rather than
    trusting the id, so an entry stranded any other way cannot wedge a key
    permanently -- it is dropped and a fresh fetch starts.

  - One callback raising no longer prevents the others being delivered.
    Previously there was only ever one.

Interaction with #499, whichever merges second: that PR releases the payload
after the callback runs. With several callbacks the release must happen
after ALL of them, and must not happen at all if a joined submitter passed
no callback, since polling get_result() would then be its only delivery
path. The callback list built here is the hook for that.

test_background_fetch_dedupe.py -- 8 tests, covering the join, callback
delivery to both submitters, one callback raising, distinct keys not being
coalesced, a post-completion submit fetching again, cancellation releasing
the key, a stranded entry not wedging one, and the reported count. Verified
non-vacuous by removing only the join branch: 3 fail. The callback tests
assert the ids coalesced, without which they would pass trivially on two
independent requests.

Full suite: 3698 passed, 60 skipped, 1 failure that reproduces identically
on unmodified main (test_install_lowmem, environment-dependent: /var/tmp is
disk-backed on this machine).

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01STMbQE4YctTacQXfbYqKuW

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
+            # Call every callback: the original submitter's and any that joined
-            if request.callback:
+            # 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

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