fix(youtube-stats): stop retrying a failing API once per frame

plugins/youtube-stats/manager.py

@@ -8,6 +8,7 @@
 """
 
 import os
+import time
 from pathlib import Path
 from typing import Dict, Any, Optional
 from PIL import Image, ImageDraw, ImageFont
@@ -72,6 +73,11 @@ def _rgb(value, default):
         self.font = None
         self.youtube_logo = None
         self.last_displayed_stats: Optional[Dict[str, Any]] = None  # Track last displayed to prevent unnecessary redraws
+        # A fetch has been attempted at least once, and when. display() used to
+        # retry on an empty result, so a failing API meant a request per FRAME;
+        # these gate on "did we try" and "how long ago" instead.
+        self._has_fetched: bool = False
+        self._last_attempt: float = 0.0
         self._api_key_error: Optional[str] = None  # Set on any config/auth problem that
         # prevents fetching stats (missing/invalid API key, missing channel ID) -- display()
         # shows this on-screen instead of leaving a blank canvas when it's set.
@@ -325,19 +331,41 @@ def _create_display(self, channel_stats: Dict[str, Any]) -> Optional[Image.Image
             return None
 
     def update(self) -> None:
-        """Fetch/update data for this plugin."""
+        """Fetch/update data for this plugin.
+
+        Throttled to update_interval. The core already calls this on roughly
+        that cadence, so this changes nothing for it -- it exists because
+        display() also calls update() for the first paint, and a failing fetch
+        leaves channel_stats empty. Without the throttle that combination
+        issued one YouTube API request per rendered frame, on the render
+        thread, each able to block for the full 10s timeout. The Data API's
+        default quota is 10k units/day, so a broken key burned through it in
+        minutes -- and quota exhaustion is itself an error, which kept the
+        result empty and the loop fed.
+
+        monotonic(), not time(): these Pis have no RTC, so a wall-clock jump
+        at NTP sync must not make the next attempt look due (or centuries away).
+        """
         if not self.enabled:
             return
-
+
+        now = time.monotonic()
+        if self._has_fetched and (now - self._last_attempt) < self.update_interval_config:
+            return
+        self._has_fetched = True
+        self._last_attempt = now
+
         self.channel_stats = self._get_channel_stats()
 
     def display(self, force_clear: bool = False) -> None:
         """Render this plugin's display."""
         if not self.enabled:
             return
 
-        # Fetch stats if we don't have them yet
-        if not self.channel_stats:
+        # First paint can land before the core's first update() tick. Gate on
+        # whether a fetch has been ATTEMPTED, not on whether it produced data:
+        # gating on the result meant a failing API was retried every frame.
+        if not self._has_fetched:
             self.update()
 
         if self.channel_stats:

plugins/youtube-stats/manifest.json

@@ -1,7 +1,7 @@
 {
   "id": "youtube-stats",
   "name": "YouTube Stats",
-  "version": "1.1.0",
+  "version": "1.1.1",
   "author": "ChuckBuilds",
   "description": "Display YouTube channel statistics including subscriber count, total views, and channel name on your LED matrix",
   "category": "social",
@@ -33,6 +33,12 @@
     }
   ],
   "versions": [
+    {
+      "version": "1.1.1",
+      "released": "2026-08-22",
+      "ledmatrix_min_version": "2.0.0",
+      "notes": "Stop retrying a failing YouTube API once per rendered frame. display() re-ran update() whenever channel_stats was empty, and update() leaves it empty on any failure, so a bad key or an exhausted quota produced one requests.get(timeout=10) per frame on the render thread. display() now gates on whether a fetch was attempted rather than on its result, and update() is throttled to update_interval, so a broken API costs one request per interval instead of one per frame."
+    },
     {
       "version": "1.1.0",
       "released": "2026-07-31",

plugins/youtube-stats/test_no_per_frame_api_storm.py

@@ -0,0 +1,120 @@
+#!/usr/bin/env python3
+"""
+Regression test: a failing YouTube API must not be retried once per frame.
+
+display() used to call update() whenever channel_stats was empty:
+
+    if not self.channel_stats:
+        self.update()
+
+update() assigns the fetch result unconditionally, so any failure -- missing
+key, HTTP error, quota exhaustion, network down -- left channel_stats empty and
+the next frame tried again. That is one requests.get(timeout=10) per rendered
+frame, on the render thread, against an API whose default quota is 10k
+units/day. Quota exhaustion is itself an error, so the loop fed itself.
+
+display() now gates on whether a fetch was ATTEMPTED, and update() is throttled
+to update_interval, so a broken API costs one request per interval.
+
+Run: <core-venv>/bin/python plugins/youtube-stats/test_no_per_frame_api_storm.py
+"""
+
+import sys
+import types
+from pathlib import Path
+
+plugin_dir = Path(__file__).parent
+sys.path.insert(0, str(plugin_dir))
+
+
+def _stub_core_src():
+    def mod(name, **attrs):
+        m = types.ModuleType(name)
+        for k, v in attrs.items():
+            setattr(m, k, v)
+        sys.modules.setdefault(name, m)
+        return m
+
+    mod("src")
+    mod("src.plugin_system")
+    mod("src.plugin_system.base_plugin", BasePlugin=object, VegasDisplayMode=object)
+
+
+_stub_core_src()
+
+import logging  # noqa: E402
+import time  # noqa: E402
+
+from manager import YouTubeStatsPlugin  # noqa: E402
+
+results = []
+
+
+def check(case, passed):
+    results.append((case, passed))
+    print(f"  [{'pass' if passed else 'FAIL'}] {case}")
+
+
+def make_plugin(update_interval=300):
+    # object.__new__ rather than YouTubeStatsPlugin.__new__: the stub above
+    # makes BasePlugin `object`, so the latter resolves to object.__new__ and
+    # static analysis reads the explicit class argument as a bound-method call
+    # with a missing `cls`. Same result, no false positive.
+    p = object.__new__(YouTubeStatsPlugin)
+    p.logger = logging.getLogger("test-youtube-storm")
+    p.enabled = True
+    p.channel_stats = None
+    p.update_interval_config = update_interval
+    p._has_fetched = False
+    p._last_attempt = 0.0
+    p._api_key_error = None
+    return p
+
+
+# --- a permanently failing API -------------------------------------------
+plugin = make_plugin()
+calls = []
+plugin._get_channel_stats = lambda: calls.append(1) or None
+
+# The core's own tick, then a burst of frames.
+plugin.update()
+for _ in range(200):
+    if not plugin._has_fetched:
+        plugin.update()
+
+check("a failing fetch is attempted once per interval, not once per frame",
+      len(calls) == 1)
+check("the failure is recorded as attempted", plugin._has_fetched is True)
+check("channel_stats stays empty (no fabricated data)", not plugin.channel_stats)
+
+# --- the throttle expires -------------------------------------------------
+plugin._last_attempt = time.monotonic() - 301
+plugin.update()
+check("a retry happens once the interval has elapsed", len(calls) == 2)
+
+# --- the throttle does not block a successful first fetch ----------------
+ok = make_plugin()
+ok_calls = []
+ok._get_channel_stats = lambda: ok_calls.append(1) or {"subscribers": 5}
+ok.update()
+check("a working API still populates stats", ok.channel_stats == {"subscribers": 5})
+check("and cost exactly one call", len(ok_calls) == 1)
+
+# --- the guard display() uses ---------------------------------------------
+import inspect  # noqa: E402
+disp = inspect.getsource(YouTubeStatsPlugin.display)
+check("display() no longer gates on the result",
+      "if not self.channel_stats:\n            self.update()" not in disp)
+check("display() gates on whether a fetch was attempted",
+      "self._has_fetched" in disp)
+
+upd = inspect.getsource(YouTubeStatsPlugin.update)
+check("update() uses monotonic (these Pis have no RTC)", "time.monotonic()" in upd)
+
+print()
+failed = [case for case, passed in results if not passed]
+print(f"{len(results) - len(failed)}/{len(results)} passed")
+if failed:
+    for case in failed:
+        print(f"  FAILED: {case}")
+    sys.exit(1)