feat(test-tools): t8n file streaming optimizations (#2751)

* feat(test-tools): file-based t8n stream; replacing in-memory stdout

Replaces Python's stdout-buffered, multi-pass parse of t8n output
(which held up to ~5 copies of the alloc in flight: raw stdout bytes,
parsed dict, re-serialized string, re-parsed dict, validated Alloc)
with a file-based LazyAllocFile that streams {address: account_dict}
entries incrementally via ``ijson`` and validates each ``Account`` one
at a time, preserving every existing Pydantic validator.

This cuts per-test peak RSS on the unchunkified benchmark from
5.38 GB → 3.40 GB (measured, fixtures byte-identical) — ~2 GB of
Python-heap pressure removed per xdist worker.

* fix(test-tools): fix issues with evm-dump-dir and ijson updates

* fix: updates from comments on PR #2751

* feat(test-tools): stream chained-block t8n input via file path

* fix(test-client-clis): ensure output dir exists in evm-dump script

* fix(test-tools): force GC in `LazyAllocFile` keepalive test under PyPy

`test_lazy_alloc_file_keepalive_pins_temp_dir` asserted that `del lazy`
immediately wipes the temp directory pinned by `LazyAllocFile._keepalive`.
That holds on CPython, where reference-counted finalization runs
`TemporaryDirectory.__del__` -> `cleanup()` synchronously when the last
reference drops. PyPy uses a generational garbage collector with no
refcounting, so after `del lazy` the `TemporaryDirectory` is unreachable
but not yet finalized, and the directory is still on disk when the next
assertion runs. Add an explicit `gc.collect()` to trigger the finalizer
deterministically on both interpreters.

* fix(test-client-clis): release `LazyAllocFile` keepalive on `OutputCache` insert

The PR introducing the file-based t8n alloc streaming path pins each producing call's `TemporaryDirectory` onto the resulting `LazyAllocFile` via `_keepalive`, so the next chained block can read `output/alloc.json` directly without re-serialization. That lifetime is correct for the live, unchained-handoff path: the predecessor's temp dir is dropped as soon as the consumer call returns and the previous `TransitionToolOutput` falls out of scope.

`OutputCache`, however, keeps every `TransitionToolOutput` produced during a single test alive for the duration of that test (single-key cache, cleared on key change). When chained-block tests run with caching enabled, every cached subcall retains its own `output/alloc.json` plus the entire surrounding `TemporaryDirectory` on `/tmp`. This is `O(N)` for an `N`-block chained test, where each `alloc.json` can be hundreds of MB to several GB; for the unchunkified benchmark scenario this could easily exhaust `/tmp` on shared CI runners.

Materialize the streamed alloc into the cached `Alloc` and clear `_keepalive` before storing the result. By the time we cache, the chained handoff has already happened (the consumer block has finished its `subprocess.run`), so the on-disk file is no longer needed for zero-copy chaining: the only future readers are cache replays from another fixture format, which want a parsed `Alloc` anyway.

Tradeoff:

- Cache replays now pay the `ijson` parse cost on insert rather than on first `.get()`. This is the same parse the live path performs lazily; doing it eagerly at cache-set is fine because we know the caller has just finished using the result, so amortizing the parse here avoids a later surprise during replay.
- The cached entry retains the parsed `Alloc` in Python heap rather than the `alloc.json` on disk. For a typical chained benchmark this is a strict improvement: one parsed `Alloc` is smaller than the equivalent JSON text plus the `TemporaryDirectory` overhead, and the peak-RSS savings of the original streaming PR are preserved during the live run (the materialization happens after the producing call has already released its working buffers).
- The streaming benefit during the first run is fully preserved; only the cached-replay path materializes.

A move-the-keepalive-to-the-consumer alternative was considered and rejected: the consumer block currently sees only the predecessor's `Path`, not the `LazyAllocFile` instance, so threading explicit cleanup back to the predecessor would be invasive and tangled with the `TransitionToolData` flow. Releasing at cache-insertion is the lowest-blast-radius point: it's the moment we know the file isn't needed for chained handoff anymore, and it's localized to `OutputCache.set` rather than spread across the two `_evaluate_*` paths.

* fix(test-client-clis): reject non-object top-level JSON in `LazyAllocFile.validate`

`ijson.kvitems(f, "")` only yields key-value pairs when the top-level JSON value is an object: for `null`, `[]`, scalars, or other valid-but-non-object inputs, the iterator silently yields zero pairs. With the streaming alloc parser landed in this branch, that means `Alloc.model_validate({})` quietly succeeds and the t8n caller receives an empty post-state instead of an error.

The pre-streaming path (`LazyAllocStr.validate`, which calls `Alloc.model_validate_json`) raised on those same inputs, so the streaming PR is a regression in error fidelity even though success-path behavior is unchanged: the t8n binaries we drive (geth, evmone, execution-specs) all produce object-shaped `alloc.json` by contract, so this failure mode does not fire under normal operation. The risk is the failure shape: a t8n that crashes after redirecting stdout, an output path that gets overwritten with an error JSON, or any future divergence from the contract would silently zero the post-state, and the test would only fail downstream with a misleading consensus mismatch rather than a clean "alloc.json is malformed" at the point of corruption. Silent corruption that surfaces far from its source is the costly kind of debugging.

Probe the first parse event from `ijson.parse` and raise `ValueError` if it is not `start_map`, then seek back to zero and let `ijson.kvitems` consume the stream as before.

Tradeoffs:

- One extra parse event per call. `ijson.parse` is event-driven; pulling a single event then re-seeking is effectively free relative to the streamed body parse, and we keep the `kvitems`-based hot path so the entry-by-entry validation pattern is unchanged.
- `ValueError` rather than reusing `ijson.IncompleteJSONError`: the input here is well-formed JSON (just the wrong shape), so reporting it as an incomplete-stream error would itself be misleading. The existing malformed-JSON tests continue to surface as `IncompleteJSONError` from inside `ijson.parse` because those inputs raise before we ever read a complete first event: the new guard only takes effect for valid JSON of the wrong type.
- The legitimately empty alloc (`{}`) is preserved: the guard accepts `start_map` and the rest of the function returns an empty `Alloc` exactly as before. New regression test covers this.

Three new parametrized cases (`null`, `[]`, `42`) cover the silent-zero scenarios, and a positive test for `{}` pins the empty-object behavior so future tightening cannot accidentally reject it.

---------

Co-authored-by: danceratopz <danceratopz@gmail.com>

Author: fselmo

packages/testing/pyproject.toml

@@ -52,6 +52,7 @@ dependencies = [
     "ckzg>=2.1.3,<3",
     "tenacity>=9.0.0,<10",
     "Jinja2>=3,<4",
+    "ijson>=3.3,<4",
 ]
 
 [project.urls]

packages/testing/src/execution_testing/client_clis/cli_types.py

@@ -1,7 +1,9 @@
 """Types used in the transition tool interactions."""
 
 import json
-from dataclasses import dataclass
+import shutil
+import tempfile
+from dataclasses import dataclass, field
 from pathlib import Path
 from typing import (
     Annotated,
@@ -10,13 +12,16 @@
     Generic,
     List,
     NamedTuple,
+    Optional,
     Self,
     TypeVar,
 )
 
+import ijson  # type: ignore[import-untyped]
 from pydantic import Field, PlainSerializer, PlainValidator
 
 from execution_testing.base_types import (
+    Account,
     Bloom,
     Bytes,
     CamelModel,
@@ -458,6 +463,55 @@ def validate(self) -> Alloc:
         return Alloc.model_validate_json(self.raw)
 
 
+@dataclass(kw_only=True)
+class LazyAllocFile(LazyAlloc[Path]):
+    """
+    Lazy allocation backed by a filesystem path.
+
+    Parses `{address: account_or_null}` entries from the file incrementally
+    via `ijson.kvitems` and validates each `Account` one at a time through
+    `Account.model_validate`. The full mapping of validated accounts is still
+    accumulated before `Alloc.model_validate` is called, so peak memory
+    scales with the size of the alloc — but the raw JSON string is never
+    held in Python memory, and there is no re-serialize / re-parse round
+    trip, which is where `LazyAllocStr` incurs its multi-GB peak.
+
+    The optional ``_keepalive`` field holds the producing t8n call's
+    ``TemporaryDirectory`` so the on-disk alloc.json survives until this
+    LazyAllocFile is dropped. That lets a chained next-block t8n call
+    consume the alloc directly from disk (via ``--input.alloc=<path>`` for
+    geth, or ``shutil.copyfile`` for filesystem t8ns) without round-tripping
+    through ``Alloc.get().model_dump_json()`` in Python.
+    """
+
+    _keepalive: Optional[tempfile.TemporaryDirectory] = field(default=None)
+
+    def validate(self) -> Alloc:
+        """Validate the alloc by streaming entries from the backing file."""
+        accumulated: Dict[str, Account | None] = {}
+        with open(self.raw, "rb") as f:
+            # `ijson.kvitems(f, "")` silently yields nothing for non-object
+            # top-level JSON (`null`, `[]`, scalars), which would turn a
+            # corrupted alloc.json into an empty post-state. Probe the first
+            # parse event so the streaming path matches the fail-loud
+            # behavior of `LazyAllocStr.validate` /
+            # `Alloc.model_validate_json`.
+            first = next(ijson.parse(f), None)
+            if first is None or first[1] != "start_map":
+                raise ValueError(
+                    f"Expected JSON object at top level of {self.raw}"
+                )
+            f.seek(0)
+            for address_str, account_data in ijson.kvitems(f, ""):
+                if account_data is None:
+                    accumulated[address_str] = None
+                else:
+                    accumulated[address_str] = Account.model_validate(
+                        account_data
+                    )
+        return Alloc.model_validate(accumulated)
+
+
 @dataclass
 class TransitionToolInput:
     """Transition tool input."""
@@ -473,13 +527,20 @@ def to_files(
         """
         Prepare the input in a directory path in the file system for
         consumption by the t8n tool.
+
+        For ``LazyAllocFile`` inputs whose backing file is still on disk
+        (chained-block handoff: previous t8n call's temp dir is pinned via
+        the keepalive field), the alloc is copied byte-for-byte rather than
+        round-tripped through ``Alloc.get().model_dump_json()``.
         """
-        if isinstance(self.alloc, Alloc):
-            alloc_contents = self.alloc.model_dump_json(**model_dump_config)
-        elif isinstance(self.alloc, LazyAllocStr):
-            alloc_contents = self.alloc.raw
+        alloc_path = directory_path / "alloc.json"
+        if (
+            isinstance(self.alloc, LazyAllocFile)
+            and Path(self.alloc.raw).exists()
+        ):
+            shutil.copyfile(self.alloc.raw, alloc_path)
         else:
-            raise Exception(f"Invalid alloc type: {type(self.alloc)}")
+            alloc_path.write_text(self._serialize_alloc(**model_dump_config))
 
         env_contents = self.env.model_dump_json(**model_dump_config)
         txs_contents = (
@@ -489,33 +550,41 @@ def to_files(
             )
             + "]"
         )
-        input_contents: Dict[str, str] = {
-            "alloc": alloc_contents,
-            "env": env_contents,
-            "txs": txs_contents,
-        }
-        if self.blob_params is not None:
-            input_contents["blobParams"] = self.blob_params.model_dump_json(
-                **model_dump_config
-            )
 
-        input_paths: Dict[str, str] = {}
-        for content_type, contents in input_contents.items():
-            file_path = directory_path / f"{content_type}.json"
+        input_paths: Dict[str, str] = {"alloc": str(alloc_path)}
+        for name, contents in (("env", env_contents), ("txs", txs_contents)):
+            file_path = directory_path / f"{name}.json"
             file_path.write_text(contents)
-            input_paths[content_type] = str(file_path)
+            input_paths[name] = str(file_path)
+        if self.blob_params is not None:
+            blob_path = directory_path / "blobParams.json"
+            blob_path.write_text(
+                self.blob_params.model_dump_json(**model_dump_config)
+            )
+            input_paths["blobParams"] = str(blob_path)
 
         return input_paths
 
-    def model_dump_json(self, **model_dump_config: Any) -> str:
-        """Dump the model in string JSON format."""
+    def _serialize_alloc(self, **model_dump_config: Any) -> str:
+        """Serialize ``self.alloc`` to a JSON string."""
         if isinstance(self.alloc, Alloc):
-            alloc_contents = self.alloc.model_dump_json(**model_dump_config)
-        elif isinstance(self.alloc, LazyAllocStr):
-            alloc_contents = self.alloc.raw
-        else:
-            raise Exception(f"Invalid alloc type: {type(self.alloc)}")
+            return self.alloc.model_dump_json(**model_dump_config)
+        if isinstance(self.alloc, LazyAllocStr):
+            return self.alloc.raw
+        if isinstance(self.alloc, LazyAllocFile):
+            return self.alloc.get().model_dump_json(**model_dump_config)
+        raise Exception(f"Invalid alloc type: {type(self.alloc)}")
+
+    def model_dump_json(
+        self, *, exclude_alloc: bool = False, **model_dump_config: Any
+    ) -> str:
+        """
+        Dump the model in string JSON format.
 
+        Pass ``exclude_alloc=True`` when the t8n is reading the alloc from a
+        file path instead of the stdin bundle, to avoid building a multi-GB
+        JSON string in Python memory for chained-block handoffs.
+        """
         env_contents = self.env.model_dump_json(**model_dump_config)
         txs_contents = (
             "["
@@ -524,18 +593,21 @@ def model_dump_json(self, **model_dump_config: Any) -> str:
             )
             + "]"
         )
-        input_contents: Dict[str, str] = {
-            "alloc": alloc_contents,
-            "env": env_contents,
-            "txs": txs_contents,
-        }
+        input_contents: Dict[str, str] = {}
+        if not exclude_alloc:
+            input_contents["alloc"] = self._serialize_alloc(
+                **model_dump_config
+            )
+        input_contents["env"] = env_contents
+        input_contents["txs"] = txs_contents
         if self.blob_params is not None:
             input_contents["blobParams"] = self.blob_params.model_dump_json(
                 **model_dump_config
             )
-        contents: List[str] = []
-        for content_type, type_contents in input_contents.items():
-            contents.append(f'"{content_type}": {type_contents}')
+        contents: List[str] = [
+            f'"{content_type}": {type_contents}'
+            for content_type, type_contents in input_contents.items()
+        ]
         return "{" + ",".join(contents) + "}"
 
     def model_dump(self, mode: str, **model_dump_config: Any) -> Any:
@@ -547,6 +619,10 @@ def model_dump(self, mode: str, **model_dump_config: Any) -> Any:
             )
         elif isinstance(self.alloc, LazyAllocJson):
             alloc_contents = self.alloc.raw
+        elif isinstance(self.alloc, LazyAllocFile):
+            alloc_contents = self.alloc.get().model_dump(
+                mode=mode, **model_dump_config
+            )
         else:
             raise Exception(f"Invalid alloc type: {type(self.alloc)}")
 
@@ -582,13 +658,19 @@ def model_validate_files(
         """
         Validate the model from the file system where each key is a
         different JSON file.
+
+        `alloc.json` is referenced by path and parsed incrementally on
+        `.get()` via `LazyAllocFile`, so the full file is never held in
+        memory alongside the validated `Alloc`.
         """
-        alloc_data = (directory_path / "alloc.json").read_text()
         result_data = (directory_path / "result.json").read_text()
         result = Result.model_validate_json(
             json_data=result_data, context=context
         )
-        alloc = LazyAllocStr(raw=alloc_data, _state_root=result.state_root)
+        alloc = LazyAllocFile(
+            raw=directory_path / "alloc.json",
+            _state_root=result.state_root,
+        )
         output = cls(result=result, alloc=alloc)
         return output
 

packages/testing/src/execution_testing/client_clis/file_utils.py

@@ -1,6 +1,7 @@
 """Methods to work with the filesystem and json."""
 
 import os
+import shutil
 import stat
 from json import dump
 from pathlib import Path
@@ -9,6 +10,7 @@
 from pydantic import BaseModel, RootModel
 
 from execution_testing.client_clis.cli_types import (
+    LazyAllocFile,
     LazyAllocJson,
     LazyAllocStr,
     TransitionToolInput,
@@ -28,27 +30,41 @@ def dump_files_to_directory(output_path: Path, files: Dict[str, Any]) -> None:
         if rel_path:
             os.makedirs(output_path / rel_path, exist_ok=True)
         file_path = output_path / file_rel_path
-        with open(file_path, "w") as f:
-            if isinstance(file_contents, (LazyAllocStr, LazyAllocJson)):
-                if isinstance(file_contents, LazyAllocJson):
-                    dump(file_contents.raw, f, ensure_ascii=True, indent=4)
-                else:
-                    f.write(file_contents.raw)
-
-            elif isinstance(
-                file_contents, (BaseModel, RootModel, TransitionToolInput)
-            ):
-                f.write(
-                    file_contents.model_dump_json(
-                        indent=4,
-                        exclude_none=True,
-                        by_alias=True,
-                    )
+        if (
+            isinstance(file_contents, LazyAllocFile)
+            and Path(file_contents.raw).exists()
+        ):
+            shutil.copyfile(file_contents.raw, file_path)
+        elif isinstance(file_contents, LazyAllocFile):
+            # Backing temp dir was cleaned up after a previous `.get()`
+            # (e.g. chained-block t8n on the next block); fall back to
+            # the cached Alloc so debug dumps still capture the input.
+            file_path.write_text(
+                file_contents.get().model_dump_json(
+                    indent=4, exclude_none=True, by_alias=True
                 )
-            elif isinstance(file_contents, str):
-                f.write(file_contents)
-            else:
-                dump(file_contents, f, ensure_ascii=True, indent=4)
+            )
+        else:
+            with open(file_path, "w") as f:
+                if isinstance(file_contents, (LazyAllocStr, LazyAllocJson)):
+                    if isinstance(file_contents, LazyAllocJson):
+                        dump(file_contents.raw, f, ensure_ascii=True, indent=4)
+                    else:
+                        f.write(file_contents.raw)
+                elif isinstance(
+                    file_contents, (BaseModel, RootModel, TransitionToolInput)
+                ):
+                    f.write(
+                        file_contents.model_dump_json(
+                            indent=4,
+                            exclude_none=True,
+                            by_alias=True,
+                        )
+                    )
+                elif isinstance(file_contents, str):
+                    f.write(file_contents)
+                else:
+                    dump(file_contents, f, ensure_ascii=True, indent=4)
         if flags:
             file_mode = os.stat(file_path).st_mode
             if "x" in flags: