refactor: audit batch — drop dead helpers, fields, properties (leanEthereum#757)

* refactor: audit batch — drop dead helpers, fields, properties

Eleven cleanups from the consensus-researcher audit, bundled.

snappy: get_uncompressed_length and is_valid_compressed_data were
tests-only helpers. The two corruption tests that used them are
rewritten to assert decompression outcomes directly.

subspecs/ssz/hash: the _htr_bytes / _htr_bytearray / _htr_memoryview
register overloads existed only for one parametrize test; both leave
together. Production hash_tree_root callers always pass typed SSZ
containers.

subspecs/xmss/rand: Rand.rho was the only Randomness-shaped helper
used solely in test_message_hash; the test now builds Randomness
from rand.field_elements directly.

subspecs/xmss/constants: XmssConfig.PUBLIC_KEY_LEN_BYTES had a
single test consumer that effectively re-derived the public-key SSZ
shape. Test rewritten to compute the expected size inline as
(HASH_LEN_FE + PARAMETER_LEN) * P_BYTES.

subspecs/observability: get_observer was only read by tests; the
private singleton is now read via observer_module._observer in the
two affected tests. NullObserver is privatised to _NullObserver and
dropped from the package's public re-exports.

subspecs/genesis/config: num_validators field and its consistency
validator were both informational; the actual count is always taken
from len(genesis_validators). from_yaml was a tests-only string
shortcut for model_validate(yaml.safe_load(...)); tests use a local
_load helper for the same ergonomics without a public method.

subspecs/validator/service: blocks_skipped_lag,
attestations_skipped_lag, and duty_gate_closed were read only from
tests. The private state machine stays; tests now access the
underscored attributes directly.

subspecs/sync/block_cache: PendingBlock.received_at was a debug-only
timestamp written but never read in src. Field dropped together with
the dedicated default-timestamp test; remaining test constructions
no longer pass received_at.

subspecs/sync/head_sync: HeadSyncResult slimmed to processed plus
error. The cached, backfill_triggered, and descendants_processed
fields had no readers outside their own assignments. The error
field is retained per the audit note for incident-response logs.
Six construction sites in head_sync.py and ~14 test assertions
updated accordingly. Also drops the unused peer_id parameter on
_process_cached_descendants; the rest of the file keeps peer_id
where it actually drives attribution.

subspecs/api/server: ApiServerConfig.enabled flag dropped (Node
already gates construction). ApiServer.run no longer polls with
asyncio.sleep(1); it awaits a stop event that _async_stop sets,
matching the same pattern adopted earlier in live.py.

Items deliberately skipped:

- Container.to_hex / from_hex: both used in xmss/containers.py.
- IntFieldElement protocol: removing requires Fp to grow its own
  Pydantic core schema; deferred as a real redesign.
- The 7 Spec*Type protocols in forks/protocol.py: serve as
  documentation labels distinguishing block-body from block-header
  from aggregated-attestations etc. Collapsing to type[SpecSSZType]
  erased meaningful type intent.
- ValidatorRegistry.primary_index: two production callers.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* fixup: apply ruff format + drop HeadSyncResult fields in routing tests

Two follow-ups to land the audit batch through CI.

- ruff format reformatted two source files (decompress.py and the
  validator service) whose edits left awkward blank-line spacing.
- A second test module, test_head_sync_backfill_routing.py, also
  constructed HeadSyncResult with the now-dropped cached,
  backfill_triggered, and descendants_processed fields. Stripped
  those argument lines so ty check passes.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* fixup: restore _htr_bytes / _htr_bytearray / _htr_memoryview overloads

The audit listed these as tests-only and the unit tests passed
without them, but the consensus fixture filler hit
"TypeError: hash_tree_root: unsupported value type bytes" the
moment a lstar State container with a raw-bytes-typed field was
walked. Production traversal does reach raw bytes; the three
overloads are load-bearing for the fixture pipeline.

Restored along with the parametrize-over-byte-likes test that was
retired together with them. Verified by filling a previously
failing fc test_tick_system fixture.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: tcoratger

src/lean_spec/snappy/__init__.py

@@ -14,8 +14,6 @@
 from .decompress import (
     SnappyDecompressionError,
     decompress,
-    get_uncompressed_length,
-    is_valid_compressed_data,
 )
 from .framing import frame_compress, frame_decompress
 
@@ -28,8 +26,6 @@
     "frame_decompress",
     # Utilities
     "max_compressed_length",
-    "get_uncompressed_length",
-    "is_valid_compressed_data",
     # Exceptions
     "SnappyDecompressionError",
 ]

src/lean_spec/snappy/decompress.py

@@ -263,64 +263,3 @@ def _execute_copy(output: bytearray, offset: int, length: int, max_length: int)
     for _ in range(length):
         output.append(output[src_pos])
         src_pos += 1
-
-
-def get_uncompressed_length(data: bytes) -> int:
-    """Read the uncompressed length from compressed data without decompressing.
-
-    Args:
-        data: Snappy-compressed bytes.
-
-    Returns:
-        The declared uncompressed length.
-
-    Raises:
-        SnappyDecompressionError: If the length varint is malformed.
-
-    Useful for:
-      - Pre-allocating buffers before decompression.
-      - Quick validation of compressed data.
-      - Checking if you have enough memory for decompression.
-    """
-    if not data:
-        raise SnappyDecompressionError("Empty input")
-
-    try:
-        length, _ = decode_varint32(data, 0)
-        return length
-    except ValueError as e:
-        raise SnappyDecompressionError(f"Invalid length varint: {e}") from e
-
-
-def is_valid_compressed_data(data: bytes) -> bool:
-    """Check if data appears to be valid Snappy-compressed data.
-
-    Args:
-        data: Data to check.
-
-    Returns:
-        True if the data appears to be valid Snappy format.
-
-    This performs quick validation WITHOUT full decompression:
-      1. Checks that the length varint is valid.
-      2. Verifies there's data after the varint (unless length is 0).
-
-    Use this for fast rejection of obviously invalid data.
-
-    Note:
-        This does NOT guarantee the data will decompress successfully.
-        It only checks the header. The compressed content may still
-        be corrupted.
-    """
-    if not data:
-        return False
-
-    try:
-        length, varint_bytes = decode_varint32(data, 0)
-
-        # Sanity checks:
-        #   - If length = 0, data is valid (empty original).
-        #   - If length > 0, there must be compressed data after the varint.
-        return length == 0 or varint_bytes < len(data)
-    except ValueError:
-        return False

src/lean_spec/subspecs/api/server.py

@@ -45,9 +45,6 @@ class ApiServerConfig:
     port: int = 5052
     """Port to listen on."""
 
-    enabled: bool = True
-    """Whether the API server is enabled."""
-
 
 @dataclass(slots=True)
 class ApiServer:
@@ -85,17 +82,16 @@ class ApiServer:
     _site: web.TCPSite | None = field(default=None, init=False)
     """TCP site for the server."""
 
+    _stop_event: asyncio.Event = field(default_factory=asyncio.Event, init=False)
+    """Set when the server stops, so run() can return without polling."""
+
     @property
     def store(self) -> Store | None:
         """Get the current Store instance."""
         return self.store_getter() if self.store_getter else None
 
     async def start(self) -> None:
         """Start the API server in the background."""
-        if not self.config.enabled:
-            logger.info("API server is disabled")
-            return
-
         app = web.Application()
 
         # Store the store_getter in app for handlers that need store access
@@ -126,9 +122,7 @@ async def run(self) -> None:
         Blocks until stop() is called.
         """
         await self.start()
-
-        while self._runner is not None:
-            await asyncio.sleep(1)
+        await self._stop_event.wait()
 
     def stop(self) -> None:
         """Request graceful shutdown (fire-and-forget). Prefer aclose() in async code."""
@@ -146,3 +140,4 @@ async def _async_stop(self) -> None:
             self._runner = None
             self._site = None
             logger.info("API server stopped")
+        self._stop_event.set()

src/lean_spec/subspecs/genesis/config.py

@@ -18,7 +18,7 @@
 from typing import Any
 
 import yaml
-from pydantic import Field, field_validator, model_validator
+from pydantic import Field, field_validator
 
 from lean_spec.forks import Validator, Validators
 from lean_spec.types import Bytes52, StrictBaseModel, Uint64, ValidatorIndex
@@ -79,14 +79,6 @@ class GenesisConfig(StrictBaseModel):
     Immutable once the chain launches.
     """
 
-    num_validators: Uint64 | None = Field(default=None, alias="NUM_VALIDATORS")
-    """
-    Number of validators (optional).
-
-    This field is informational and may be included in config files.
-    The actual validator count is derived from the genesis validator list.
-    """
-
     genesis_validators: list[GenesisValidatorEntry] = Field(alias="GENESIS_VALIDATORS")
     """
     Validators trusted to secure the chain from slot 0.
@@ -99,18 +91,6 @@ class GenesisConfig(StrictBaseModel):
     Security note: 2/3+ collusion controls the chain until new validators join.
     """
 
-    @model_validator(mode="after")
-    def validate_num_validators_consistency(self) -> GenesisConfig:
-        """Verify num_validators matches actual count when provided."""
-        if self.num_validators is not None:
-            actual_count = len(self.genesis_validators)
-            if int(self.num_validators) != actual_count:
-                raise ValueError(
-                    f"NUM_VALIDATORS ({self.num_validators}) does not match "
-                    f"actual validator count ({actual_count})"
-                )
-        return self
-
     def to_validators(self) -> Validators:
         """
         Build the genesis validator set with assigned indices.
@@ -146,13 +126,3 @@ def from_yaml_file(cls, path: Path | str) -> GenesisConfig:
         with path.open(encoding="utf-8") as f:
             data = yaml.safe_load(f)
         return cls.model_validate(data)
-
-    @classmethod
-    def from_yaml(cls, content: str) -> GenesisConfig:
-        """
-        Load configuration from a YAML string.
-
-        Useful for testing or programmatic config generation.
-        """
-        data = yaml.safe_load(content)
-        return cls.model_validate(data)

src/lean_spec/subspecs/observability/__init__.py

@@ -12,19 +12,15 @@
 """
 
 from .observer import (
-    NullObserver,
     SpecObserver,
-    get_observer,
     observe_on_attestation,
     observe_on_block,
     observe_state_transition,
     set_observer,
 )
 
 __all__ = [
-    "NullObserver",
     "SpecObserver",
-    "get_observer",
     "observe_on_attestation",
     "observe_on_block",
     "observe_state_transition",

src/lean_spec/subspecs/observability/observer.py

@@ -42,7 +42,7 @@ def on_attestation_timed(self, seconds: float) -> None:
         """Report the wall time of validating and integrating a gossip attestation."""
 
 
-class NullObserver:
+class _NullObserver:
     """
     Default observer that discards every event.
 
@@ -60,11 +60,11 @@ def on_attestation_timed(self, seconds: float) -> None:  # noqa: ARG002
         """Accept and discard."""
 
 
-_observer: SpecObserver = NullObserver()
+_observer: SpecObserver = _NullObserver()
 """
 Process-wide observer singleton.
 
-Starts as a NullObserver so spec imports are side-effect-free.
+Starts as a _NullObserver so spec imports are side-effect-free.
 Replaced by the client at startup via set_observer.
 """
 
@@ -81,16 +81,6 @@ def set_observer(observer: SpecObserver) -> None:
     _observer = observer
 
 
-def get_observer() -> SpecObserver:
-    """
-    Return the currently registered observer.
-
-    Spec code calls this to publish events.
-    When no observer has been registered the returned value is a NullObserver.
-    """
-    return _observer
-
-
 @contextmanager
 def observe_state_transition() -> Iterator[None]:
     """

src/lean_spec/subspecs/sync/block_cache.py

@@ -41,7 +41,6 @@
 
 from collections import OrderedDict, defaultdict
 from dataclasses import dataclass, field
-from time import time
 
 from lean_spec.forks import SignedBlock, Store
 from lean_spec.subspecs.networking.transport.peer_id import PeerId
@@ -110,15 +109,6 @@ class PendingBlock:
     None for self-produced blocks.
     """
 
-    received_at: float = field(default_factory=time)
-    """
-    Unix timestamp when the block was received.
-
-    Enables staleness detection and debugging.
-
-    Very old pending blocks may indicate a stuck backfill or network issues.
-    """
-
     backfill_depth: int = 0
     """
     Depth of backfill chain from original request.

src/lean_spec/subspecs/sync/head_sync.py

@@ -64,22 +64,13 @@ class HeadSyncResult:
     """
     Result of processing a gossip block.
 
-    Provides detailed feedback about what happened when a block was received.
-    This allows the SyncService to make informed decisions about state.
+    The SyncService only branches on the processed flag; the error string is
+    retained for incident-response logs and post-mortem inspection.
     """
 
     processed: bool
     """True if the block was immediately integrated into the Store."""
 
-    cached: bool
-    """True if the block was added to the cache (parent unknown)."""
-
-    backfill_triggered: bool
-    """True if backfill was initiated for missing parents."""
-
-    descendants_processed: int
-    """Number of cached descendants that were also processed."""
-
     error: str | None = None
     """Error message if processing failed."""
 
@@ -177,19 +168,13 @@ async def on_gossip_block(
             logger.debug("on_gossip_block: skipping - already processing")
             return HeadSyncResult(
                 processed=False,
-                cached=False,
-                backfill_triggered=False,
-                descendants_processed=0,
             ), store
 
         # Skip if already in store (duplicate).
         if block_root in store.blocks:
             logger.debug("on_gossip_block: skipping - already in store")
             return HeadSyncResult(
                 processed=False,
-                cached=False,
-                backfill_triggered=False,
-                descendants_processed=0,
             ), store
 
         # Check if parent exists in store.
@@ -255,24 +240,17 @@ async def _process_block_with_descendants(
                 )
                 return HeadSyncResult(
                     processed=False,
-                    cached=False,
-                    backfill_triggered=False,
-                    descendants_processed=0,
                     error=str(e),
                 ), store
 
             # Process cached descendants.
-            descendants_count, store = await self._process_cached_descendants(
+            _, store = await self._process_cached_descendants(
                 parent_root=block_root,
                 store=store,
-                peer_id=peer_id,
             )
 
             return HeadSyncResult(
                 processed=True,
-                cached=False,
-                backfill_triggered=False,
-                descendants_processed=descendants_count,
             ), store
 
         finally:
@@ -282,7 +260,6 @@ async def _process_cached_descendants(
         self,
         parent_root: Bytes32,
         store: Store,
-        peer_id: PeerId | None,
     ) -> tuple[int, Store]:
         """
         Process any cached blocks that descend from the given parent.
@@ -296,7 +273,6 @@ async def _process_cached_descendants(
         Args:
             parent_root: Root of the parent block just processed.
             store: Current store (may be updated during processing).
-            peer_id: Peer ID for error attribution.
 
         Returns:
             Tuple of (descendants successfully processed, updated store).
@@ -331,7 +307,6 @@ async def _process_cached_descendants(
                     desc_count, store = await self._process_cached_descendants(
                         parent_root=child_root,
                         store=store,
-                        peer_id=peer_id,
                     )
                     processed_count += desc_count
 
@@ -382,9 +357,6 @@ async def _cache_and_backfill(
             )
             return HeadSyncResult(
                 processed=False,
-                cached=False,
-                backfill_triggered=False,
-                descendants_processed=0,
             ), store
 
         # Add to cache.
@@ -427,9 +399,6 @@ async def _cache_and_backfill(
 
         return HeadSyncResult(
             processed=False,
-            cached=True,
-            backfill_triggered=True,
-            descendants_processed=0,
         ), store
 
     def reset(self) -> None: