mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-08-25 20:38:15 +00:00
fix(memory): release fetched payloads once they have been delivered (#499)
* fix(memory): release fetched payloads once they have been delivered BackgroundDataService kept the fetched body on the FetchResult it filed in completed_requests, which is swept hourly and capped at 500 entries by count. For a status record that costs nothing; for a season schedule it costs a tenth of the board. Measured on a 1GB Pi 3B+ with a 1-second RSS profile: the display process sat at 404MB after plugin load, then stepped +21MB when NFL fetched its season and +90MB when NCAA football fetched 946 games for 2026 -- and stayed at 494MB. Not a leak; a staircase that never came down. When a later fetch landed while headroom was low, available memory reached ~70MB, fork() began failing, and the board stopped being able to start a process at all: sshd accepted connections and closed them before its banner, systemd could not respawn the display, and the panel went dark while the kernel carried on answering pings. The cache-hit path was the worse of the two. It runs once per update interval per sport, mints a fresh request_id each time, and files whatever the cache returned. The memory tier is capped at 150 entries on a 1GB board, so a miss re-parses the payload from disk into a genuinely new object -- separate copies accumulating toward the 500-entry cap, not shared references. Releasing is safe: the payload is written to the cache under the request's cache_key before the result is built, the callback is handed the object directly, and consumers read it back from the cache afterwards (the plugins' callbacks use it only in passing, to log a count, before reading the cache). Nothing is lost -- it moves from RAM to the disk cache that was already holding it. Requests submitted without a callback keep their payload, since polling get_result() is then the only way to collect it. That keeps the existing contract, and the existing tests covering it, intact. Not addressed here: max_workers=3 allows three concurrent fetches, so three large parses can peak at once, and there is no in-flight dedupe by cache_key -- a second submit for a key already being fetched starts a second fetch. Both bound the transient peak rather than what stays resident, and both are behaviour changes worth their own review. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * fix(memory): file the cache-hit result before running its callback Restores the original ordering. Releasing the payload after the callback meant filing the result after it too, so a callback that queried get_result() or is_request_complete() for its own request would not have found it -- a behaviour change unrelated to the memory fix. The dict holds a reference to the same object, so releasing after filing still clears the payload. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * test: wait for the payload release, not just the filing The callback test waited on is_request_complete(), which goes true as soon as the worker files the result in completed_requests. The worker then runs the cleanup pass, then the callback, then releases the payload. Both of the test's assertions therefore raced the worker: `seen` is populated by the callback, and `data is None` only after the release that follows it. It passes today because a one-line callback usually finishes inside the 20ms poll interval. Confirmed by making the callback sleep 0.4s: _wait() returns with seen == {} and the payload still resident. _wait_for_release() polls for the released payload instead. Release happens strictly after the callback returns, so a released payload also means the callback has finished and one wait covers both assertions. Verified against the same 0.4s callback. _wait() stays for the other three fetch-path tests, which assert only what is already true when the result is filed -- the success flag, the error, and the cache write that happened during the fetch itself. Its docstring now says so, so the next reader picks the right one. 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 <noreply@anthropic.com>
This commit is contained in:
Chuck
@@ -57,7 +57,15 @@ class FetchRequest:
|
||||
|
||||
@dataclass
|
||||
class FetchResult:
|
||||
"""Result of a background fetch operation."""
|
||||
"""Result of a background fetch operation.
|
||||
|
||||
``data`` survives on the stored result only for requests submitted without
|
||||
a ``callback``, where polling ``get_result()`` is the sole way to collect
|
||||
it. When a callback was given, the payload has already been delivered and
|
||||
the service releases it -- see :meth:`BackgroundDataService._release_payload`.
|
||||
Either way the data remains in the cache under the request's ``cache_key``,
|
||||
which is where consumers read it from.
|
||||
"""
|
||||
request_id: str
|
||||
success: bool
|
||||
data: Optional[Any] = None
|
||||
@@ -191,14 +199,19 @@ class BackgroundDataService:
|
||||
cached=True,
|
||||
fetch_time=0.0
|
||||
)
|
||||
# Filed before the callback runs, as it always was: a callback
|
||||
# that queries get_result()/is_request_complete() for its own
|
||||
# request must still find it. Releasing afterwards mutates the
|
||||
# same object the dict holds.
|
||||
self.completed_requests[request_id] = result
|
||||
|
||||
|
||||
if callback:
|
||||
try:
|
||||
callback(result)
|
||||
except Exception as e:
|
||||
logger.error(f"Error in callback for request {request_id}: {e}")
|
||||
|
||||
self._release_payload(result)
|
||||
|
||||
logger.debug(f"Cache hit for {sport} {year} data")
|
||||
return request_id
|
||||
|
||||
@@ -333,8 +346,29 @@ class BackgroundDataService:
|
||||
request.callback(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
|
||||
# object, so one survivor keeps the whole payload resident.
|
||||
self._release_payload(result)
|
||||
request.result = None
|
||||
|
||||
return result
|
||||
|
||||
@staticmethod
|
||||
def _release_payload(result: FetchResult) -> None:
|
||||
"""Drop a delivered payload, keeping the result's status and timings.
|
||||
|
||||
Only called once a callback has been handed the data. Consumers read
|
||||
fetched data back from the cache under ``cache_key``; the copy carried
|
||||
here was pinning a parsed season schedule -- 946 games for NCAA
|
||||
football, roughly a tenth of total RAM on a 1GB Pi -- in memory until
|
||||
the hourly sweep.
|
||||
|
||||
The cache-hit path matters most: it runs once per update interval per
|
||||
sport, mints a fresh request_id each time, and a memory-tier miss
|
||||
re-parses the payload from disk. Those were genuinely separate copies
|
||||
accumulating toward the 500-entry cap, not shared references.
|
||||
"""
|
||||
result.data = None
|
||||
|
||||
def _make_request_with_retry(self, request: FetchRequest) -> requests.Response:
|
||||
"""
|
||||
|
||||
@@ -0,0 +1,187 @@
|
||||
"""A delivered fetch payload must not stay resident on the stored result.
|
||||
|
||||
BackgroundDataService kept the fetched body on the FetchResult it filed in
|
||||
`completed_requests`, which is swept only hourly and capped at 500 entries by
|
||||
count. For status records that is free; for a season schedule it is not. NCAA
|
||||
football's 2026 schedule is 946 games, and on a 1GB Pi 3B+ the parsed payload
|
||||
measured ~90MB -- a tenth of the board's memory, pinned for an hour after the
|
||||
consumer had already been handed it.
|
||||
|
||||
The cache-hit path was the worse of the two. It runs once per update interval
|
||||
per sport, mints a fresh request_id each time, and hands back whatever the
|
||||
cache returns -- so a memory-tier miss (the tier is capped at 150 entries)
|
||||
re-parses the payload from disk into a genuinely new object. Those accumulate
|
||||
as separate copies rather than shared references, which is the staircase seen
|
||||
in the field: RSS stepping up ~90MB per sport as seasons loaded and never
|
||||
coming back down.
|
||||
|
||||
Releasing is safe because the payload is written to the cache under the
|
||||
request's cache_key before the result is built, and that is where consumers
|
||||
read it from -- the callback is handed the object directly and the plugins use
|
||||
it only in passing before reading the cache back.
|
||||
|
||||
Requests submitted *without* a callback keep their payload: polling
|
||||
get_result() is then the only way to collect it, so releasing would break that
|
||||
contract.
|
||||
"""
|
||||
|
||||
import time
|
||||
import pytest
|
||||
from unittest.mock import MagicMock, Mock, patch
|
||||
|
||||
from src.background_data_service import BackgroundDataService
|
||||
|
||||
|
||||
PAYLOAD = {"events": [{"id": f"g{i}"} for i in range(50)]}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def cache():
|
||||
m = MagicMock()
|
||||
m.get.return_value = None
|
||||
m.set.return_value = None
|
||||
return m
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def service(cache):
|
||||
svc = BackgroundDataService(cache, max_workers=2, request_timeout=5)
|
||||
yield svc
|
||||
svc.shutdown(wait=False)
|
||||
|
||||
|
||||
def _wait(service, req_id, timeout=5):
|
||||
"""Wait for the result to be FILED.
|
||||
|
||||
Enough for anything that is true by the time the worker stores the result:
|
||||
its success flag, its error, the cache write that happened during the
|
||||
fetch.
|
||||
"""
|
||||
deadline = time.time() + timeout
|
||||
while not service.is_request_complete(req_id) and time.time() < deadline:
|
||||
time.sleep(0.02)
|
||||
|
||||
|
||||
def _wait_for_release(service, req_id, timeout=5):
|
||||
"""Wait for the payload to be RELEASED, which is strictly later.
|
||||
|
||||
The worker files the result, then runs the callback, then releases. So
|
||||
is_request_complete() goes true while the callback still has not run --
|
||||
waiting on it alone leaves a window in which `seen` is empty and the
|
||||
payload is still resident, and the assertions race the worker. It passes
|
||||
in practice only because a one-line callback usually beats the 20ms poll.
|
||||
|
||||
Release happens after the callback returns, so a released payload also
|
||||
means the callback has finished: one wait covers both.
|
||||
"""
|
||||
deadline = time.time() + timeout
|
||||
while time.time() < deadline:
|
||||
result = service.get_result(req_id)
|
||||
if result is not None and result.data is None:
|
||||
return
|
||||
time.sleep(0.02)
|
||||
raise AssertionError(
|
||||
f"payload for {req_id} was never released (callback may not have run)")
|
||||
|
||||
|
||||
def _resp():
|
||||
r = Mock()
|
||||
r.json.return_value = PAYLOAD
|
||||
r.raise_for_status.return_value = None
|
||||
return r
|
||||
|
||||
|
||||
class TestFetchPath:
|
||||
def test_callback_receives_the_payload_then_it_is_released(self, service, cache):
|
||||
seen = {}
|
||||
|
||||
def callback(result):
|
||||
# The consumer's one look at the data happens here.
|
||||
seen['events'] = len(result.data['events'])
|
||||
|
||||
with patch.object(service.session, "get", return_value=_resp()):
|
||||
req_id = service.submit_fetch_request(
|
||||
sport="ncaa_fb", year=2026, url="https://example.com/s",
|
||||
cache_key="ncaa_fb_2026", callback=callback, max_retries=0,
|
||||
)
|
||||
_wait_for_release(service, req_id)
|
||||
|
||||
assert seen['events'] == 50, "callback must still be handed the payload"
|
||||
|
||||
stored = service.get_result(req_id)
|
||||
assert stored is not None
|
||||
assert stored.success is True
|
||||
assert stored.data is None, "payload must not stay on the stored result"
|
||||
|
||||
def test_nothing_is_lost_the_cache_holds_it(self, service, cache):
|
||||
with patch.object(service.session, "get", return_value=_resp()):
|
||||
req_id = service.submit_fetch_request(
|
||||
sport="ncaa_fb", year=2026, url="https://example.com/s",
|
||||
cache_key="ncaa_fb_2026", callback=lambda r: None, max_retries=0,
|
||||
)
|
||||
_wait(service, req_id)
|
||||
|
||||
cache.set.assert_called_once()
|
||||
key, written = cache.set.call_args[0][:2]
|
||||
assert key == "ncaa_fb_2026"
|
||||
assert written == PAYLOAD, "the payload must be persisted before release"
|
||||
|
||||
def test_without_a_callback_the_payload_is_kept(self, service, cache):
|
||||
# Polling get_result() is then the only delivery mechanism.
|
||||
with patch.object(service.session, "get", return_value=_resp()):
|
||||
req_id = service.submit_fetch_request(
|
||||
sport="nfl", year=2026, url="https://example.com/s",
|
||||
cache_key="nfl_2026", max_retries=0,
|
||||
)
|
||||
_wait(service, req_id)
|
||||
|
||||
assert service.get_result(req_id).data == PAYLOAD
|
||||
|
||||
def test_a_failed_fetch_still_records_its_error(self, service, cache):
|
||||
with patch.object(service.session, "get", side_effect=Exception("boom")):
|
||||
req_id = service.submit_fetch_request(
|
||||
sport="nfl", year=2026, url="https://example.com/s",
|
||||
cache_key="nfl_2026", callback=lambda r: None, max_retries=0,
|
||||
)
|
||||
_wait(service, req_id)
|
||||
|
||||
stored = service.get_result(req_id)
|
||||
assert stored.success is False
|
||||
assert stored.error is not None
|
||||
|
||||
|
||||
class TestCacheHitPath:
|
||||
def test_cache_hit_releases_after_the_callback(self, service, cache):
|
||||
cache.get.return_value = PAYLOAD
|
||||
seen = {}
|
||||
|
||||
req_id = service.submit_fetch_request(
|
||||
sport="ncaa_fb", year=2026, url="https://example.com/s",
|
||||
cache_key="ncaa_fb_2026",
|
||||
callback=lambda r: seen.update(events=len(r.data['events'])),
|
||||
)
|
||||
|
||||
assert seen['events'] == 50
|
||||
assert service.get_result(req_id).data is None
|
||||
|
||||
def test_repeated_cache_hits_do_not_accumulate_payloads(self, service, cache):
|
||||
# The staircase: one entry per update interval per sport, each one
|
||||
# potentially a freshly parsed copy after a memory-tier miss.
|
||||
cache.get.return_value = PAYLOAD
|
||||
|
||||
for _ in range(25):
|
||||
service.submit_fetch_request(
|
||||
sport="ncaa_fb", year=2026, url="https://example.com/s",
|
||||
cache_key="ncaa_fb_2026", callback=lambda r: None,
|
||||
)
|
||||
|
||||
retained = [r for r in service.completed_requests.values() if r.data is not None]
|
||||
assert retained == [], f"{len(retained)} payloads still resident"
|
||||
|
||||
def test_cache_hit_without_a_callback_is_unchanged(self, service, cache):
|
||||
cache.get.return_value = PAYLOAD
|
||||
req_id = service.submit_fetch_request(
|
||||
sport="nfl", year=2026, url="https://example.com/s",
|
||||
cache_key="nfl_2026",
|
||||
)
|
||||
assert service.get_result(req_id).data == PAYLOAD
|
||||
Reference in New Issue
Block a user