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

* 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

* fix(memory): discard a cancelled fetch instead of letting it commit

Review follow-up on the dedupe.

Cancelling releases the cache_key, so a replacement fetch for that key can
start immediately. But _fetch_data_worker() had no cancellation check: the
cancelled worker still wrote its response to the cache, flipped its own
status from CANCELLED to COMPLETED, and ran its callbacks. The stale
response could therefore land on top of the replacement's fresher data.

The worker cannot abort an HTTP call in flight, so the response is discarded
on return instead: no cache write, no callbacks, status left CANCELLED. The
check sits immediately before the cache write, which is the first
side effect.

Also fixed, found by the new test rather than by reading:

    request_id was f"{sport}_{year}_{milliseconds}", which is not unique.
    Two submits inside the same millisecond produced the SAME id -- the
    test's two sequential fetches collided on a fast mocked response, and
    one request silently replaced the other in active_requests and
    completed_requests. Rare before this PR; load-bearing now, because
    dedupe hands that id back to every joiner as their handle for
    get_result(). A per-service counter is appended.

Two test problems of my own, both fixed here rather than left to flake:

  - The cancellation test synchronised with time.sleep(0.4). A slow worker
    would have made it pass for the wrong reason. It now waits for the
    request to be filed in completed_requests.
  - The id-uniqueness test patched session.get, but submits are async: the
    50 workers outlived the patch and made real DNS calls to the dummy host.
    It stubs the executor instead, which is what a submit-time test should
    exercise.

20 consecutive runs of the dedupe file: 0 failures. Full suite: 3700 passed,
60 skipped, 1 failure that reproduces identically on unmodified main
(test_install_lowmem, environment-dependent).

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

* fix(memory): make cancellation terminal, not advisory

Three paths wrote request.status without checking whether the request had
already been cancelled, so a cancel could be silently undone and the work it
was meant to stop went ahead anyway.

- A request cancelled while queued had CANCELLED overwritten with IN_PROGRESS
  the moment its worker started, defeating the discard check entirely: it
  downloaded, cached and called back for work the caller had withdrawn. It
  now skips the fetch outright, which is also the cheapest possible cancel.
- The cancelled-check and the cache write were separate critical sections, so
  a cancel landing between them left the payload in the cache with the
  callbacks suppressed -- every submitter that joined the fetch waited for a
  call that never came. The worker now claims the commit in the same critical
  section that reads the status, and cancel_request refuses once claimed. The
  write stays outside the lock: it serialises a multi-megabyte payload to the
  SD card, and holding the service lock across that would stall every submit,
  status query and cancel behind it.
- A cancelled request that then failed was relabelled FAILED, which slipped
  past the CANCELLED-only callback gate and delivered a spurious error
  callback. The except path now leaves CANCELLED alone.

get_request_status() also reported a cancelled request as FAILED, since it
inferred status from result.success; the final status is now recorded on the
result. Both early returns assign to `result` so completed_requests files the
outcome that was reported rather than the untouched placeholder.

Tests cover cancellation before worker start, during the commit, and during an
HTTP failure; all three fail against the unfixed code.

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

* test(memory): reach the exception path as a cancelled request

test_a_failure_after_cancelling_stays_cancelled cancelled the request while
its worker was still queued, so once the pre-start branch landed the worker
returned there and never reached the exception handler the test is named for.
It passed against the unfixed code only because that branch did not exist yet;
with it, the test passed for the wrong reason and reverting the except-path
guard did not fail it.

Cancel while the worker is parked inside the HTTP call instead, and assert the
fetch actually started so the test cannot silently degrade into the pre-start
case again. Reverting each of the three guards now fails exactly one test.

Also read the payload inside the callback rather than off the FetchResult
afterwards: #499 releases result.data once delivery is done, so the later read
saw the released object and not what the caller was handed.

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

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

test/test_background_fetch_dedupe.py

@@ -0,0 +1,450 @@
+"""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
+import requests
+
+from src.background_data_service import BackgroundDataService, FetchStatus
+
+
+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
+
+
+def test_a_cancelled_worker_cannot_overwrite_its_replacement(service, cache):
+    """Cancelling frees the key, so a replacement may already own it.
+
+    The worker cannot abort an HTTP call in flight, so when the cancelled one
+    finally returns it must discard its response rather than write it. Without
+    that, the sequence is: cancel A, submit B for the same key, B fetches and
+    caches fresh data, A returns and overwrites it with the response nobody
+    wanted -- and calls A's callbacks too.
+    """
+    slow = _BlockingSession()
+    stale = {"events": [{"id": "STALE"}]}
+    slow_resp = Mock()
+    slow_resp.json.return_value = stale
+    slow_resp.raise_for_status.return_value = None
+
+    def blocked_get(*a, **k):
+        slow.calls += 1
+        slow.started.set()
+        slow.release.wait(timeout=5)
+        return slow_resp
+
+    called = []
+    with patch.object(service.session, "get", side_effect=blocked_get):
+        first = service.submit_fetch_request(
+            sport="nba", year=2026, url="https://x/s", cache_key="k",
+            callback=lambda r: called.append("cancelled_one"), max_retries=0)
+        assert slow.started.wait(timeout=5)
+
+        service.cancel_request(first)
+        assert "k" not in service._inflight_by_cache_key
+
+    # The replacement writes the fresh value while the cancelled fetch is held.
+    fresh = {"events": [{"id": "FRESH"}]}
+    fresh_resp = Mock()
+    fresh_resp.json.return_value = fresh
+    fresh_resp.raise_for_status.return_value = None
+    with patch.object(service.session, "get", return_value=fresh_resp):
+        second = service.submit_fetch_request(
+            sport="nba", year=2026, url="https://x/s", cache_key="k",
+            callback=lambda r: called.append("replacement"), max_retries=0)
+        _wait(service, second)
+
+    assert cache.set.call_args[0][1] == fresh, "replacement must own the cache"
+
+    # Now let the cancelled fetch finish. It must write nothing and call nobody.
+    # Wait for the worker to actually finish rather than sleeping: a fixed
+    # sleep is a race under load, and a slow worker would make this pass for
+    # the wrong reason. A cancelled request is still filed in
+    # completed_requests, so that is the signal it has run to completion.
+    writes_before = cache.set.call_count
+    slow.release.set()
+    deadline = time.time() + 5
+    while first not in service.completed_requests and time.time() < deadline:
+        time.sleep(0.02)
+    assert first in service.completed_requests, "cancelled worker never finished"
+
+    assert cache.set.call_count == writes_before, (
+        "the cancelled worker wrote to the cache after its replacement")
+    assert cache.set.call_args[0][1] == fresh, "stale data overwrote fresh"
+    assert "cancelled_one" not in called, (
+        "a cancelled request must not deliver callbacks")
+
+
+def test_request_ids_are_unique_within_a_millisecond(service):
+    """request_id was sport_year_milliseconds, which collides.
+
+    Two submits inside the same millisecond produced the SAME id, so one
+    silently replaced the other in active_requests and completed_requests.
+    Dedupe hands this id back to every joiner as their handle for
+    get_result(), so uniqueness is now load-bearing rather than incidental.
+    """
+    # Stub the executor rather than the session: this is about what submit
+    # hands back, and letting 50 workers loose would outlive the patch and
+    # make real network calls.
+    with patch.object(service.executor, "submit"):
+        ids = [
+            service.submit_fetch_request(
+                sport="nba", year=2026, url="https://x/s",
+                cache_key=f"key_{i}",           # distinct keys: no dedupe
+                callback=lambda r: None, max_retries=0)
+            for i in range(50)
+        ]
+    assert len(set(ids)) == len(ids), "request ids collided"
+
+
+# --- cancellation must be terminal ------------------------------------------
+#
+# Cancelling used to be advisory: three separate paths wrote request.status
+# without checking whether the request had already been cancelled, so a cancel
+# could be silently undone and the work it was meant to stop went ahead.
+
+URL = "http://example.invalid/scores"
+KEY = "sched_nfl_2025"
+
+
+class _CountingSession:
+    """Records whether an HTTP fetch was ever attempted."""
+
+    def __init__(self):
+        self.calls = 0
+
+    def get(self, *a, **k):
+        self.calls += 1
+        return _resp()
+
+
+class _BlockingFailingSession(_CountingSession):
+    """Holds the fetch open, then fails it.
+
+    Cancelling while the worker is parked inside the HTTP call is the only way
+    to reach the exception handler as a cancelled request. Cancel it before
+    the call starts and the worker returns at the pre-start branch instead,
+    which would leave the except path untested.
+    """
+
+    def __init__(self):
+        super().__init__()
+        self.started = threading.Event()
+        self.release = threading.Event()
+
+    def get(self, *a, **k):
+        self.calls += 1
+        self.started.set()
+        self.release.wait(timeout=5)
+        raise requests.RequestException("connection reset")
+
+
+def _fill_every_worker_slot(service, slots=3):
+    """Occupy the pool so the next submit is queued rather than started.
+
+    This is what makes "cancel before the worker runs" deterministic instead
+    of a race the test would win only sometimes. Returns the gate that
+    releases the pool.
+    """
+    gate = threading.Event()
+    for _ in range(slots):
+        service.executor.submit(gate.wait, 5)
+    return gate
+
+
+def test_cancelling_before_the_worker_starts_stops_the_fetch(service, cache):
+    """The queued worker must honour a cancel, not overwrite it with IN_PROGRESS.
+
+    Between submit and the worker picking the job up, the request sits in the
+    executor queue. Cancelling there is the cheapest possible cancel -- nothing
+    has been downloaded yet -- and it was the one that did not work.
+    """
+    gate = _fill_every_worker_slot(service)
+    session = _CountingSession()
+    delivered = []
+
+    with patch.object(service, 'session', session):
+        rid = service.submit_fetch_request(
+            "nfl", 2025, URL, KEY, max_retries=0, callback=delivered.append
+        )
+        assert service.cancel_request(rid) is True
+        gate.set()                       # let the queued worker run
+        _wait(service, rid)
+
+    assert session.calls == 0, (
+        "cancelled before it started, yet the worker still downloaded the payload"
+    )
+    assert cache.set.call_count == 0, "a cancelled request wrote to the cache"
+    assert delivered == [], "a cancelled request invoked its callbacks"
+
+
+def test_a_cancel_during_the_commit_is_refused(service, cache):
+    """Once the worker has claimed the commit, cancelling is too late.
+
+    The claim and the cancelled-check happen in one critical section, so a
+    cancel arriving after it cannot retroactively abandon data already on its
+    way to the cache. Letting it through stranded every joiner: the payload
+    landed in the cache but the callbacks were suppressed, so a manager that
+    joined this fetch waited for a call that never came.
+    """
+    gate = _fill_every_worker_slot(service)
+    late = {}
+    delivered = []
+
+    def cancel_mid_write(key, data, *a, **k):
+        late['returned'] = service.cancel_request(late['rid'])
+
+    cache.set.side_effect = cancel_mid_write
+
+    # Read the payload inside the callback. The service releases result.data
+    # once every callback has been delivered, so inspecting the FetchResult
+    # afterwards sees the released object, not what the caller was handed.
+    def record(result):
+        delivered.append((result.success, result.data))
+
+    with patch.object(service, 'session', _CountingSession()):
+        late['rid'] = service.submit_fetch_request(
+            "nfl", 2025, URL, KEY, max_retries=0, callback=record
+        )
+        gate.set()                       # only now can the worker reach the commit
+        _wait(service, late['rid'])
+
+    assert late.get('returned') is False, (
+        "cancelled a request that had already committed"
+    )
+    assert cache.set.call_count == 1, "the commit itself was lost"
+    assert delivered == [(True, PAYLOAD)], (
+        "data reached the cache but the callbacks were suppressed -- "
+        "every joined submitter is left waiting forever"
+    )
+
+
+def test_a_failure_after_cancelling_stays_cancelled(service, cache):
+    """A cancelled request that then errors must not resurface as FAILED.
+
+    The except path overwrote CANCELLED with FAILED, and the callback gate in
+    the finally block only suppresses callbacks for CANCELLED -- so cancelling
+    a request that was about to time out delivered a spurious error callback.
+    """
+    session = _BlockingFailingSession()
+    delivered = []
+
+    with patch.object(service, 'session', session):
+        rid = service.submit_fetch_request(
+            "nfl", 2025, URL, KEY, max_retries=0, callback=delivered.append
+        )
+        # Assert the worker is inside the HTTP call before cancelling,
+        # otherwise this silently degrades into the pre-start case and the
+        # exception handler is never exercised.
+        assert session.started.wait(timeout=5)
+        assert service.cancel_request(rid) is True
+        session.release.set()
+        _wait(service, rid)
+
+    assert session.calls == 1, "the fetch never started, so nothing could fail"
+
+    assert delivered == [], "a cancelled request delivered a failure callback"
+    assert service.get_request_status(rid) is FetchStatus.CANCELLED, (
+        "a cancelled request that then errored was reported as FAILED"
+    )