fix(validator): fall back to last-known-good PAT snapshot on transient read failure

gittensor/validator/pat_storage.py

@@ -29,29 +29,47 @@
 )
 
 _lock = threading.Lock()
+_last_known_good_pats: Optional[list[dict]] = None
+
+
+def _snapshot_entries(entries: list[dict]) -> list[dict]:
+    """Return a defensive copy and remember it as the last successful read."""
+    global _last_known_good_pats
+    copied = [entry.copy() for entry in entries]
+    _last_known_good_pats = copied
+    return copied
 
 
 def ensure_pats_file() -> None:
     """Create the PATs file with an empty list if it doesn't exist. Called on validator boot."""
     with _lock:
         if not PATS_FILE.exists():
             _write_file([])
+            _snapshot_entries([])
 
 
 def load_all_pats() -> list[dict]:
     """Snapshot all stored PAT entries for a scoring round.
 
     Read-only and deliberately tolerant: an unreadable store here must not crash
-    the round (an unhandled error would stop the validator) nor wipe anything. It
-    logs loudly and returns [] so the round recovers on the next successful read.
-    The *write* path (save_pat) is the one that fails closed.
+    the round (an unhandled error would stop the validator) nor wipe anything.
+    On a transient read failure, returns the last successful snapshot so miners
+    are not all treated as PAT-less for the round. The *write* path (save_pat)
+    is the one that fails closed.
     """
     with _lock:
         try:
-            return _read_file()
+            return _snapshot_entries(_read_file())
         except (json.JSONDecodeError, OSError) as e:
+            if _last_known_good_pats is not None:
+                bt.logging.warning(
+                    f'miner_pats.json unreadable this round; using last known good PAT snapshot '
+                    f'({len(_last_known_good_pats)} miners) until the store recovers: {e}'
+                )
+                return [entry.copy() for entry in _last_known_good_pats]
             bt.logging.error(
-                f'miner_pats.json unreadable this round; scoring with no stored PATs until it recovers: {e}'
+                f'miner_pats.json unreadable and no prior PAT snapshot exists; '
+                f'scoring with no stored PATs until it recovers: {e}'
             )
             return []
 
@@ -83,6 +101,7 @@ def save_pat(uid: int, hotkey: str, pat: str, github_id: str) -> None:
             entries.append(entry)
 
         _write_file(entries)
+        _snapshot_entries(entries)
 
 
 def get_pat_by_uid(uid: int) -> Optional[dict]:

tests/validator/test_pat_storage.py

@@ -17,6 +17,7 @@ def use_tmp_pats_file(tmp_path, monkeypatch):
     """Redirect PAT storage to a temporary file for each test."""
     tmp_file = tmp_path / 'miner_pats.json'
     monkeypatch.setattr(pat_storage, 'PATS_FILE', tmp_file)
+    monkeypatch.setattr(pat_storage, '_last_known_good_pats', None)
     return tmp_file
 
 
@@ -86,11 +87,23 @@ def test_load_returns_all_entries(self):
         entries = pat_storage.load_all_pats()
         assert len(entries) == 2
 
-    def test_load_handles_corrupt_file(self, use_tmp_pats_file):
+    def test_load_handles_corrupt_file_without_prior_snapshot(self, use_tmp_pats_file):
         use_tmp_pats_file.write_text('not json{{{')
         entries = pat_storage.load_all_pats()
         assert entries == []
 
+    def test_load_corrupt_file_falls_back_to_last_snapshot(self, use_tmp_pats_file):
+        pat_storage.save_pat(1, 'h1', 'p1', 'user_1')
+        pat_storage.save_pat(2, 'h2', 'p2', 'user_2')
+        assert {e['uid'] for e in pat_storage.load_all_pats()} == {1, 2}
+
+        use_tmp_pats_file.write_text('not json{{{')
+        entries = pat_storage.load_all_pats()
+        assert len(entries) == 2
+        assert {e['uid'] for e in entries} == {1, 2}
+        assert entries[0]['pat'] == 'p1'
+        assert entries[1]['pat'] == 'p2'
+
 
 class TestSavePatFailsClosed:
     """The read-then-overwrite wipe (issue #1481, Proof 2): a single failed read of