🐛 fix(cache): write compact interpreter JSON

docs/changelog/204.bugfix.rst

@@ -0,0 +1 @@
+Write interpreter cache entries as compact JSON to keep their size proportional to the interpreter response.

src/python_discovery/_cache.py

@@ -105,6 +105,10 @@ def _read_entry(self) -> tuple[dict | None, bool]:
             ):
                 _LOGGER.warning("ignoring untrusted cache entry at %s", self._file)
                 return None, True
+            # A 1 MiB query can expand to 12 MiB when supplementary Unicode is escaped.
+            if status.st_size > 16 << 20:
+                _LOGGER.warning("ignoring oversized cache entry at %s", self._file)
+                return None, True
             return json.load(file), False
 
     @staticmethod
@@ -119,7 +123,8 @@ def write(self, content: dict) -> None:
                 mode="w", encoding="utf-8", dir=self._folder, prefix=f".{self._key}.", delete=False
             ) as file:
                 temporary = Path(file.name)
-                json.dump(content, file, sort_keys=True, indent=2)
+                # Compact encoding avoids indentation growth and the streaming encoder's recursion limit.
+                file.write(json.dumps(content, sort_keys=True))
             temporary.replace(self._file)
         except OSError:
             _LOGGER.debug("failed to write %s", self._file, exc_info=True)

tests/test_cache.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import json
 import os
 import stat
 import sys
@@ -36,6 +37,40 @@ def test_disk_content_store_read_missing_file(tmp_path: Path) -> None:
     assert store.read() is None
 
 
+def test_disk_content_store_reads_near_limit_escaped_json(tmp_path: Path) -> None:
+    content = {"padding": "😀" * ((1 << 20) - 4_096)}
+    store = DiskContentStore(tmp_path, "test")
+    store.write(content)
+    assert store.read() == content
+
+
+def test_disk_content_store_discards_oversized_entry(tmp_path: Path) -> None:
+    entry = tmp_path / "test.json"
+    entry.write_text(json.dumps({"padding": "x" * (16 << 20)}), encoding="utf-8")
+    store = DiskContentStore(tmp_path, "test")
+    assert store.read() is None
+    assert not entry.exists()
+
+
+@pytest.mark.parametrize(
+    "source",
+    [
+        pytest.param(json.dumps("x" * 398), id="flat"),
+        pytest.param("[" * 200 + "]" * 200, id="nested"),
+    ],
+)
+def test_disk_content_store_write_size_tracks_input(tmp_path: Path, source: str) -> None:
+    content = {"padding": json.loads(source)}
+    DiskContentStore(tmp_path, "test").write(content)
+    assert (tmp_path / "test.json").stat().st_size < 2 * len(json.dumps(content))
+
+
+def test_disk_content_store_write_deep_json(tmp_path: Path) -> None:
+    depth = 2_000 if sys.version_info >= (3, 15) else 900
+    DiskContentStore(tmp_path, "test").write({"padding": json.loads("[" * depth + "]" * depth)})
+    assert (tmp_path / "test.json").stat().st_size < 10_000
+
+
 def test_disk_content_store_remove(tmp_path: Path) -> None:
     store = DiskContentStore(tmp_path, "test")
     store.write({"key": "value"})