Compare commits

...
3 Commits
Author SHA1 Message Date
ChuckandClaude Opus 5 eae063700f test(sports): pin the four-case sunset matrix before B6 runs (#505)
B5 and B6 make different promises about an adopted scoreboard, and only the
second is obvious. The table:

                     | bundled copy present   | bundled copy removed
  pinned old core    | loads (the fallback)   | ERROR, naming the module
  current core       | loads, using core code | loads, using core code

The top-left cell is the one worth having. Nothing in the suite proved that an
adopted plugin still runs on a core predating src.common.sports_scroll, and
that claim is the entire basis for having shipped B5 ahead of B6's gate.

The bottom-left cell is asserted through PluginManager.load_plugin rather than
a bare import, deliberately. The manager catches the ModuleNotFoundError, so a
test written around pytest.raises would pass against a core where the module is
merely broken rather than absent, and would say nothing about what the user
meets: a plugin parked in ERROR and one log line. The assertion is the ERROR
state plus an error naming the module, which is what makes the log actionable.

Modelling notes, both of which were wrong first time and matter:

- The copy-removed shape is an UNGUARDED import, not the guarded one with the
  legacy file deleted. Keeping the guard while removing its fallback only
  mislabels the failure -- the plugin reports a missing scroll_display_legacy
  and never mentions the core module that is actually absent.
- Only the leaf module is hidden. A pre-3.2.0 core still ships src/common/;
  hiding the package would be a harsher core than any that shipped, and would
  make the failure name the package instead of the module.

Ablation: disabling the old-core simulation fails three cells, and dropping the
error object from load_plugin's set_state fails the fourth, so none of it
passes vacuously. The install gate -- the other half of the guarantee -- stays
in test_plugin_compatibility_gate.py rather than being duplicated here.

Refs docs/SPORTS_UNIFICATION.md, "B6 -- why the sunset needs more than a
version floor".


Claude-Session: https://claude.ai/code/session_01STMbQE4YctTacQXfbYqKuW

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-26 11:25:10 -04:00
ChuckandClaude Opus 5 af96bd5cb0 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>
2026-08-26 10:03:43 -04:00
Chuck 5e5979973e Update README.md (#504)
Signed-off-by: Chuck <33324927+ChuckBuilds@users.noreply.github.com>
2026-08-26 09:13:23 -04:00
4 changed files with 919 additions and 9 deletions
+4 -2
View File
@@ -161,9 +161,11 @@ The system supports live, recent, and upcoming game information for multiple spo
### LED Matrix Panels
(2x in a horizontal chain is recommended)
- [Adafruit 64×32](https://www.adafruit.com/product/2278) – designed for 128×32 but works with dynamic scaling on many displays (pixel pitch is user preference)
**Warning: Lately the Waveshare Panels have had different variations - only some are compatible with this project. I hope to identify what is different to fix it but so far there is a decent chance you get a mis-matched set of panels if you don't buy them all at once! **
- [Waveshare 64×32](https://amzn.to/3Kw55jK) - Does not require E addressable pad
- [Waveshare 96×48](https://amzn.to/4bydNcv) – higher resolution, requires soldering the **E addressable pad** on the [Adafruit RGB Bonnet](https://www.adafruit.com/product/3211) to “8” **OR** toggling the DIP switch on the Adafruit Triple LED Matrix Bonnet *(no soldering required!)*
> Amazon Affiliate Link – ChuckBuilds receives a small commission on purchases
- [Waveshare 96×48](https://amzn.to/4bydNcv) – higher resolution, requires soldering the **E addressable pad** on the [Adafruit RGB Bonnet](https://www.adafruit.com/product/3211) to “8” **OR** toggling the DIP switch on the Adafruit Triple LED Matrix Bonnet *(no soldering required!)*
- There are some Panels on Aliexpress that have worked fine for me, shop around! I think Adafruit is probably the "safest" but they do have some limitation on resolution and layout.
> Amazon Affiliate Links – ChuckBuilds receives a small commission on purchases
### Power Supply
- [5V 4A DC Power Supply](https://www.adafruit.com/product/658) (good for 2 -3 displays, depending on brightness and pixel density, you'll need higher amperage for more)
+155 -7
View File
@@ -14,11 +14,12 @@ Key Features:
- Memory-efficient data storage
"""
import itertools
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,8 +51,19 @@ 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
# Set once the worker has decided this response will be cached, while it
# still holds the lock. From that point cancelling is refused: the write
# is already authorised, and abandoning it here would put the payload in
# the cache with the callbacks suppressed -- joiners waiting forever for a
# fetch that did, in fact, succeed.
commit_claimed: bool = False
result: Optional[Any] = None
error: Optional[str] = None
@@ -74,6 +86,10 @@ class FetchResult:
fetch_time: float = 0.0
retry_count: int = 0
completed_at: float = field(default_factory=time.time) # Timestamp when request completed
# The request's final status, recorded so a finished request can still be
# reported accurately. Without it a caller can only be told COMPLETED or
# FAILED, which turns "you cancelled this" into "this errored".
final_status: Optional[FetchStatus] = None
class BackgroundDataService:
"""
@@ -98,6 +114,20 @@ 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] = {}
# 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()
@@ -185,7 +215,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)
@@ -231,7 +263,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
@@ -256,7 +310,31 @@ class BackgroundDataService:
try:
with self._lock:
request.status = FetchStatus.IN_PROGRESS
# A request cancelled while it sat in the executor queue must
# stay cancelled. Overwriting the status here undid the cancel
# outright: the worker went on to download, cache and call back
# for work the caller had already withdrawn.
if request.status == FetchStatus.CANCELLED:
cancelled_before_start = True
else:
cancelled_before_start = False
request.status = FetchStatus.IN_PROGRESS
if cancelled_before_start:
logger.info(
"Request %s was cancelled before its worker started; "
"skipping the fetch entirely", request.id
)
# Assign before returning: the finally block stores `result`
# in completed_requests, so building a fresh one here would
# file the untouched placeholder instead of this outcome.
result = FetchResult(
request_id=request.id,
success=False,
error="cancelled",
fetch_time=time.time() - start_time,
retry_count=request.retry_count
)
return result
logger.info(f"Starting background fetch for {request.sport} {request.year}")
@@ -282,6 +360,37 @@ 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 not cancelled:
# Claim the commit in the same critical section that read
# the status, so a cancel cannot slip in between the check
# and the cache write below. The write itself 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.
request.commit_claimed = True
if cancelled:
logger.info(
"Discarding response for cancelled request %s; %s may "
"already belong to a replacement fetch",
request.id, request.cache_key
)
result = FetchResult(
request_id=request.id,
success=False,
error="cancelled",
fetch_time=time.time() - start_time,
retry_count=request.retry_count
)
return result
# Cache the data
self.cache_manager.set(request.cache_key, data)
@@ -307,7 +416,12 @@ class BackgroundDataService:
logger.error(f"Failed to fetch {request.sport} {request.year} data: {error_msg}")
with self._lock:
request.status = FetchStatus.FAILED
# Don't relabel a cancelled request. The callback gate in the
# finally block only suppresses CANCELLED, so promoting it to
# FAILED here delivered an error callback for a fetch nobody
# was waiting on any more.
if request.status != FetchStatus.CANCELLED:
request.status = FetchStatus.FAILED
request.error = error_msg
result = FetchResult(
@@ -321,9 +435,26 @@ class BackgroundDataService:
finally:
# Store result and clean up
with self._lock:
result.final_status = request.status
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]
# 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:
@@ -340,10 +471,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}")
# Delivered. Drop both references -- they point at the same
@@ -456,6 +588,8 @@ class BackgroundDataService:
return self.active_requests[request_id].status
elif request_id in self.completed_requests:
result = self.completed_requests[request_id]
if result.final_status is not None:
return result.final_status
return FetchStatus.COMPLETED if result.success else FetchStatus.FAILED
return None
@@ -472,8 +606,22 @@ class BackgroundDataService:
with self._lock:
if request_id in self.active_requests:
request = self.active_requests[request_id]
if request.commit_claimed:
# Too late: the worker holds an authorised commit. Report
# the failure rather than half-cancelling a request whose
# data is about to land in the cache.
logger.debug(
"Not cancelling %s: its response is already being "
"committed", request_id
)
return False
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
+450
View File
@@ -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"
)
+310
View File
@@ -0,0 +1,310 @@
"""What happens to an adopted sports plugin on each core it can meet.
B5 moved the eight scoreboards onto `src.common.sports_scroll` behind a guarded
import, keeping a bundled copy as the fallback. B6 deletes those copies. The
two phases make *different* promises, and only the second one is obvious:
| | bundled copy present | bundled copy removed |
|--------------------|---------------------------------|-----------------------------|
| **pinned old core**| loads -- B5's whole guarantee | ERROR, naming the module |
| **current core** | loads, using core code | loads, using core code |
The top-left cell is the one worth having: nothing else in this suite proves an
adopted plugin still runs on a core that predates the module, and that claim is
the entire basis for having shipped B5 ahead of B6's gate.
The bottom-left cell is the B6 failure mode, and it is asserted through
`PluginManager.load_plugin` rather than a bare import on purpose. The manager
catches the `ModuleNotFoundError`, so nothing propagates to a caller: a test
that expected `pytest.raises` would pass against a core where the module is
merely *broken* rather than absent, and would say nothing about what the user
actually experiences. What they get is a plugin parked in ERROR and one log
line -- which is precisely why B6 needs the install gate rather than trusting
the failure to be noticed.
This covers the load path. The install/update gate -- which is what should stop
a sunset plugin reaching an old core in the first place -- is the other half of
the guarantee and is tested in test_plugin_compatibility_gate.py.
See docs/SPORTS_UNIFICATION.md, "B6 -- why the sunset needs more than a version
floor".
"""
import itertools
import json
import sys
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
project_root = Path(__file__).parent.parent
if str(project_root) not in sys.path:
sys.path.insert(0, str(project_root))
from src.plugin_system.plugin_manager import PluginManager
from src.plugin_system.plugin_state import PluginState
_PLUGIN_IDS = itertools.count()
CORE_MODULE = "src.common.sports_scroll"
# Only the leaf. A pre-3.2.0 core still ships `src/common/` -- scroll_helper
# and friends live there -- and it is `sports_scroll.py` alone that is absent.
# Hiding the whole package would be a different, harsher core than any that
# shipped, and it would make the load failure name the package rather than the
# module, which is the thing a reader needs to see.
HIDDEN = (CORE_MODULE,)
class _PinnedOldCore:
"""Make `src.common.sports_scroll` un-importable for the duration.
A meta_path finder rather than a monkeypatched `__import__`: the plugin is
executed by the real loader through `exec_module`, so the block has to live
in the import system itself to be reached.
"""
def __init__(self, *names):
self.names = set(names)
self._saved = {}
def find_spec(self, fullname, path=None, target=None):
if fullname in self.names:
raise ModuleNotFoundError(f"No module named {fullname!r}", name=fullname)
return None
def __enter__(self):
for name in list(sys.modules):
if name in self.names:
self._saved[name] = sys.modules.pop(name)
sys.meta_path.insert(0, self)
return self
def __exit__(self, *exc):
sys.meta_path.remove(self)
sys.modules.update(self._saved)
return False
# The two source shapes, kept as literals rather than copied from a plugin at
# runtime: this file lives in the core repo and must not depend on a plugin
# checkout being present, and pinning the shapes here means a plugin that
# drifts away from one is a visible edit, not a silently weakened test.
# B5, as the eight scoreboards ship today: prefer the core module, fall back to
# the bundled copy. The except clause is narrow on purpose -- a bare
# `except ImportError` would also swallow a failure raised *inside* a core
# module that is present, quietly loading the legacy copy and hiding a broken
# core install.
ADOPTED_WITH_FALLBACK = '''
_USING_CORE_SCROLL = False
try:
from src.common.sports_scroll import SportsScrollDisplay as _Base
_USING_CORE_SCROLL = True
except ModuleNotFoundError as exc:
if exc.name not in {"src", "src.common", "src.common.sports_scroll"}:
raise
_Base = None
if not _USING_CORE_SCROLL:
from scroll_display_legacy import LegacyScrollDisplay as _Base
'''
# B6, once the copies are deleted: there is nothing to fall back to, so the
# guard goes with them. Keeping the try/except while removing the file it
# falls back to would only mislabel the failure -- the plugin would report a
# missing `scroll_display_legacy` and say nothing about the core module that
# is actually absent.
SUNSET_NO_FALLBACK = '''
from src.common.sports_scroll import SportsScrollDisplay as _Base
_USING_CORE_SCROLL = True
'''
PLUGIN_BODY = '''
from src.plugin_system.base_plugin import BasePlugin
class SunsetProbe(BasePlugin):
"""Records which scroll implementation the guarded import selected."""
using_core_scroll = _USING_CORE_SCROLL
scroll_base = _Base
def update(self):
pass
def display(self, force_clear=False):
pass
'''
LEGACY_COPY = '''
class LegacyScrollDisplay:
"""Stands in for the bundled pre-3.2.0 implementation."""
'''
def _write_plugin(plugins_dir: Path, plugin_id: str, *, bundled_copy: bool) -> Path:
path = plugins_dir / plugin_id
path.mkdir(parents=True)
(path / "manifest.json").write_text(
json.dumps({
"id": plugin_id,
"name": "Sunset Probe",
"version": "1.0.0",
"entry_point": "manager.py",
"class_name": "SunsetProbe",
"display_modes": ["sunset_probe"],
}),
encoding="utf-8",
)
shape = ADOPTED_WITH_FALLBACK if bundled_copy else SUNSET_NO_FALLBACK
(path / "manager.py").write_text(shape + PLUGIN_BODY, encoding="utf-8")
if bundled_copy:
(path / "scroll_display_legacy.py").write_text(LEGACY_COPY, encoding="utf-8")
return path
@pytest.fixture
def load(tmp_path):
"""Load a synthetic adopted plugin through the real PluginManager.
Returns a callable taking the two axes of the matrix and handing back the
manager, so the caller can ask it for state and recorded error.
"""
plugins_dir = tmp_path / "plugin-repos"
plugins_dir.mkdir()
def _load(*, core_has_module: bool, bundled_copy: bool, hide=HIDDEN):
# A distinct id per cell, from a counter that spans the whole session.
# The loader names plugin modules after the plugin id and sys.modules
# is process-global, so a per-test counter would hand the second test
# the first test's already-imported module -- which passes or fails on
# the wrong plugin's import.
plugin_id = f"sunset-probe-{next(_PLUGIN_IDS)}"
_write_plugin(plugins_dir, plugin_id, bundled_copy=bundled_copy)
with patch('src.common.permission_utils.ensure_directory_permissions'):
manager = PluginManager(
plugins_dir=str(plugins_dir),
config_manager=MagicMock(),
display_manager=MagicMock(),
cache_manager=MagicMock(),
font_manager=MagicMock(),
)
manager.discover_plugins()
if core_has_module:
ok = manager.load_plugin(plugin_id)
else:
with _PinnedOldCore(*hide):
ok = manager.load_plugin(plugin_id)
return manager, plugin_id, ok
return _load
def _assert_loaded(manager, plugin_id, ok):
assert ok is True, (
f"load_plugin returned False; state is "
f"{manager.state_manager.get_state(plugin_id)}, error "
f"{manager.state_manager.get_error_info(plugin_id)}"
)
assert manager.state_manager.get_state(plugin_id) is not PluginState.ERROR
class TestBundledCopyPresent:
"""B5's shape: the guarded import with the fallback still shipped."""
def test_old_core_falls_back_and_still_loads(self, load):
"""The claim that made it safe to ship B5 before B6's gate."""
manager, plugin_id, ok = load(core_has_module=False, bundled_copy=True)
_assert_loaded(manager, plugin_id, ok)
plugin = manager.plugins[plugin_id]
assert plugin.using_core_scroll is False, (
"the core module was hidden, so the plugin must be on its bundled copy"
)
assert plugin.scroll_base.__name__ == "LegacyScrollDisplay"
def test_current_core_prefers_the_core_module(self, load):
manager, plugin_id, ok = load(core_has_module=True, bundled_copy=True)
_assert_loaded(manager, plugin_id, ok)
plugin = manager.plugins[plugin_id]
assert plugin.using_core_scroll is True, (
"the bundled copy must not win while the core module is importable"
)
assert plugin.scroll_base.__name__ == "SportsScrollDisplay"
def test_a_core_without_the_package_at_all_still_falls_back(self, load):
"""The guard's other accepted names.
Its except clause accepts `src` and `src.common` as well as the module
itself, so those branches exist in all eight shipped plugins. No core
that old is likely still running, but the code claiming to handle it is
real and nothing else exercises it -- an untested branch in a fallback
is exactly the kind that rots unnoticed until the fallback is needed.
"""
manager, plugin_id, ok = load(
core_has_module=False, bundled_copy=True, hide=(CORE_MODULE, "src.common")
)
_assert_loaded(manager, plugin_id, ok)
assert manager.plugins[plugin_id].using_core_scroll is False
class TestBundledCopyRemoved:
"""B6's shape: the copies are gone and only the core module remains."""
def test_current_core_still_loads(self, load):
manager, plugin_id, ok = load(core_has_module=True, bundled_copy=False)
_assert_loaded(manager, plugin_id, ok)
assert manager.plugins[plugin_id].using_core_scroll is True
def test_old_core_errors_and_records_the_missing_module(self, load):
"""The B6 failure mode, as the user meets it.
Not `pytest.raises`: load_plugin catches it, so nothing reaches a
caller. The observable consequences are the ERROR state and the
recorded error -- and the error has to name the module, or whoever
reads the log cannot tell a missing core module from any other
import failure.
"""
manager, plugin_id, ok = load(core_has_module=False, bundled_copy=False)
assert ok is False, "a plugin with no scroll implementation must not load"
assert manager.state_manager.get_state(plugin_id) is PluginState.ERROR
assert plugin_id not in manager.plugins, (
"a plugin that failed to load must not be left registered"
)
info = manager.state_manager.get_error_info(plugin_id)
assert info is not None, "ERROR state recorded no error to explain it"
assert info['error_type'] == 'ModuleNotFoundError', info
assert CORE_MODULE in info['error'], (
f"the recorded error must name the module that was missing, got {info['error']!r}"
)
def test_the_matrix_has_one_failing_cell(load):
"""Guards the shape of the table itself.
Each cell above is asserted on its own, so a change that broke two of them
in compensating ways could leave every individual test passing. This says
the outcome depends on both axes and fails in exactly one combination.
"""
outcomes = {
(core, bundled): load(core_has_module=core, bundled_copy=bundled)[2]
for core in (True, False)
for bundled in (True, False)
}
assert outcomes == {
(True, True): True,
(True, False): True,
(False, True): True,
(False, False): False,
}, outcomes