From 82d0bebe2e3ca7a0332b08addb54650c14f956d4 Mon Sep 17 00:00:00 2001 From: ChuckBuilds Date: Tue, 25 Aug 2026 14:19:08 -0400 Subject: [PATCH] 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) Claude-Session: https://claude.ai/code/session_01STMbQE4YctTacQXfbYqKuW --- src/background_data_service.py | 43 +++++++++++++- test/test_background_fetch_dedupe.py | 85 ++++++++++++++++++++++++++++ 2 files changed, 125 insertions(+), 3 deletions(-) diff --git a/src/background_data_service.py b/src/background_data_service.py index e6891d60..325fb82d 100644 --- a/src/background_data_service.py +++ b/src/background_data_service.py @@ -14,6 +14,7 @@ Key Features: - Memory-efficient data storage """ +import itertools import time import logging import threading @@ -103,6 +104,12 @@ class BackgroundDataService: # 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] = {} + # request_id was sport_year_milliseconds, which is not unique: two + # submits inside the same millisecond produced the SAME id, so one + # silently replaced the other in active_requests and completed_requests. + # Rare before, but dedupe hands this id back to every joiner as their + # handle for get_result(), so it has to be unique. A counter is enough. + self._request_seq = itertools.count() self.active_requests: Dict[str, FetchRequest] = {} self.completed_requests: Dict[str, FetchResult] = {} self.request_queue = queue.PriorityQueue() @@ -190,7 +197,9 @@ class BackgroundDataService: if cache_key is None: cache_key = self.get_sport_cache_key(sport) - request_id = f"{sport}_{year}_{int(time.time() * 1000)}" + with self._lock: + request_id = (f"{sport}_{year}_{int(time.time() * 1000)}" + f"_{next(self._request_seq)}") # Check cache first cached_data = self.cache_manager.get(cache_key) @@ -304,6 +313,28 @@ class BackgroundDataService: # Log data validation logger.debug(f"Validated {len(events)} events for {request.sport} {request.year}") + # A cancelled request must not commit anything. Cancelling + # releases the cache_key, so a replacement fetch for the same key + # may already be in flight or finished -- writing this response to + # the cache now would overwrite fresher data with the response + # nobody wanted. The worker has no way to abort the HTTP call, so + # this is where the work gets discarded. + with self._lock: + cancelled = request.status == FetchStatus.CANCELLED + if cancelled: + logger.info( + "Discarding response for cancelled request %s; %s may " + "already belong to a replacement fetch", + request.id, request.cache_key + ) + return FetchResult( + request_id=request.id, + success=False, + error="cancelled", + fetch_time=time.time() - start_time, + retry_count=request.retry_count + ) + # Cache the data self.cache_manager.set(request.cache_key, data) @@ -354,8 +385,14 @@ class BackgroundDataService: # 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) + # A cancelled request delivers nothing: its joiners were told + # about a fetch that has been abandoned, and a replacement will + # call them via its own request. + if request.status == FetchStatus.CANCELLED: + callbacks = [] + else: + callbacks = ([request.callback] if request.callback else []) + callbacks.extend(request.extra_callbacks) # Update statistics if result.success: diff --git a/test/test_background_fetch_dedupe.py b/test/test_background_fetch_dedupe.py index 6438c6cf..5b209d2e 100644 --- a/test/test_background_fetch_dedupe.py +++ b/test/test_background_fetch_dedupe.py @@ -210,3 +210,88 @@ def test_the_deduplicated_count_is_reported(service): 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"