refactor(sync): collapse SyncState to enum plus single non-trivial property (leanEthereum#729)

SyncState was 163 lines for a 3-state enum. Most of the surface was
trivial `state == SyncState.X` wrappers and a state-machine helper
whose only caller was sync service. Net change: -225 lines across
4 files, zero behavior change.

Changes:

- states.py drops `is_idle`, `is_syncing`, `is_synced` properties
  (trivial == wrappers), `can_transition_to` (one caller), and the
  module-level `_VALID_TRANSITIONS` table. Keeps the enum and
  `accepts_gossip`, the only property doing real work.
- service.py drops `SyncService.is_syncing` and `is_synced` public
  wrappers (zero external callers). Replaces three call sites that
  used the removed enum properties with direct comparisons.
- service.py rewrites `_transition_to` to inline the validity check.
  The original 5-edge transition table becomes two invariants:
  no self-transitions, no IDLE -> SYNCED shortcut. Same semantics
  verified case-by-case against the original table.
- test_states.py shrinks to cover the surviving surface (enum
  identity and accepts_gossip). The `_transition_to` invariants
  now have coverage in test_service.py, including a new
  self-transition test to replace the parametrized test that lived
  on `can_transition_to`.

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

Author: tcoratger

src/lean_spec/subspecs/sync/service.py

@@ -406,16 +406,6 @@ def state(self) -> SyncState:
         """Current sync state."""
         return self._state
 
-    @property
-    def is_syncing(self) -> bool:
-        """Check if actively syncing."""
-        return self._state.is_syncing
-
-    @property
-    def is_synced(self) -> bool:
-        """Check if synced with network."""
-        return self._state.is_synced
-
     def get_progress(self) -> SyncProgress:
         """
         Get current sync progress.
@@ -722,7 +712,7 @@ async def _check_sync_trigger(self) -> None:
         #
         # If already SYNCING, we should not re-trigger.
         # This prevents redundant state transitions.
-        if self._state.is_syncing:
+        if self._state == SyncState.SYNCING:
             return
 
         # Guard: Require peer information before syncing.
@@ -741,7 +731,7 @@ async def _check_sync_trigger(self) -> None:
         # network has finalized past our head, we are definitely behind.
         if network_finalized > head_slot:
             await self._transition_to(SyncState.SYNCING)
-        elif self._state.is_idle:
+        elif self._state == SyncState.IDLE:
             # Transition from IDLE even if caught up.
             #
             # IDLE -> SYNCING enables gossip processing. Even if our head matches
@@ -756,7 +746,7 @@ async def _check_sync_complete(self) -> None:
         finalized slot and there are no orphan blocks.
         """
         # Guard: Only check completion while actively syncing.
-        if not self._state.is_syncing:
+        if self._state != SyncState.SYNCING:
             return
 
         # Invariant: All orphan blocks must be resolved before declaring synced.
@@ -781,23 +771,19 @@ async def _check_sync_complete(self) -> None:
             await self._transition_to(SyncState.SYNCED)
 
     async def _transition_to(self, new_state: SyncState) -> None:
-        """
-        Transition to a new sync state.
+        """Transition to a new sync state, rejecting invalid moves.
 
-        Args:
-            new_state: Target state.
+        Two invariants are enforced:
+
+        - No self-transitions: a transition must change the current state.
+        - No IDLE -> SYNCED shortcut: SYNCING must run before SYNCED is reached.
 
-        Raises:
-            ValueError: If transition is not allowed.
+        Every other (from, to) pair is allowed, including any state -> IDLE.
         """
-        # Validate the transition against the state machine rules.
-        #
-        # The state machine enforces valid transitions:
-        # - IDLE -> SYNCING (start sync)
-        # - SYNCING -> SYNCED (caught up)
-        # - SYNCED -> SYNCING (fell behind)
-        # - Any -> IDLE (reset)
-        if not self._state.can_transition_to(new_state):
+        forbidden = new_state == self._state or (
+            self._state == SyncState.IDLE and new_state == SyncState.SYNCED
+        )
+        if forbidden:
             raise ValueError(f"Invalid state transition: {self._state.name} -> {new_state.name}")
 
         self._state = new_state

src/lean_spec/subspecs/sync/states.py

@@ -6,158 +6,30 @@
 
 
 class SyncState(Enum):
-    """
-    Sync service states representing the current synchronization phase.
-
-    This is a simple three-state machine for reactive synchronization:
-
-    State Machine Diagram
-
-    ::
-
-        IDLE --> SYNCING --> SYNCED
-          ^         |           |
-          +---------+-----------+
-
-    The Lifecycle
+    """Three-phase progression for the sync service.
 
-    A newly started node follows this progression:
+    Lifecycle:
 
-    1. **IDLE**: Node starts, no peers connected yet
-    2. **SYNCING**: Peers report chain ahead of us; react to gossip blocks
-    3. **SYNCED**: Local head reaches network finalized slot; fully synchronized
+        IDLE -> SYNCING -> SYNCED
+          ^         |         |
+          +---------+---------+
 
-    How It Works
+    - IDLE: no peers connected, or shutdown requested.
+    - SYNCING: active block processing and backfill driven by gossip.
+    - SYNCED: caught up to the network finalized slot.
 
-    - Blocks arrive via gossip
-    - If parent is known, process immediately
-    - If parent is unknown, cache block and fetch parent (backfill)
-    - Backfill happens naturally within SYNCING, not as a separate state
-
-    Transitions
-
-    IDLE -> SYNCING
-        - Triggered when: Peers connected and we need to sync
-        - Action: Start processing gossip blocks
-
-    SYNCING -> SYNCED
-        - Triggered when: local_head >= network_finalized_slot and no orphans
-        - Action: Transition to passive mode
-
-    SYNCED -> SYNCING
-        - Triggered when: Gap detected or fell behind
-        - Action: Resume active sync
-
-    Any -> IDLE
-        - Triggered when: No connected peers or shutdown requested
-        - Action: Pause all sync activity
+    Either active state may fall back to IDLE on disconnect.
+    SYNCED falls back to SYNCING when a gap reappears.
     """
 
     IDLE = auto()
-    """
-    Inactive state: no synchronization in progress.
-
-    The sync service enters IDLE when:
-
-    - **Startup**: Before any peers connect
-    - **No peers**: All peers disconnected or unreachable
-    - **Shutdown**: Graceful termination requested
-
-    While IDLE, the service waits passively. No requests are sent. The only
-    way out is connecting to peers and receiving Status messages.
-    """
-
+    """No peers connected, or shutdown requested."""
     SYNCING = auto()
-    """
-    Active synchronization state: processing gossip and backfilling.
-
-    SYNCING is the main working state. The node receives gossip blocks and
-    processes them, backfilling missing parents as needed.
-
-    In this state:
-
-    - Gossip blocks are processed immediately if parent is known
-    - Unknown parents trigger backfill requests
-    - Cached blocks are processed when parents arrive
-    """
-
+    """Active block processing and backfill driven by gossip."""
     SYNCED = auto()
-    """
-    Fully synchronized state: at or past network finalized slot.
-
-    SYNCED is the goal state. The node's head has reached or passed the
-    network's finalized checkpoint. This means:
-
-    - We have all finalized blocks
-    - We are following the chain head in real-time
-    - No active sync activity is needed
-
-    In this state:
-
-    - Gossip blocks are still processed
-    - Falls back to SYNCING if gaps appear
-    """
-
-    def can_transition_to(self, target: SyncState) -> bool:
-        """
-        Check if transition to target state is valid.
-
-        State machines enforce invariants through transition rules. This method
-        encodes those rules. Callers should check validity before transitioning
-        to catch logic errors early.
-
-        Args:
-            target: The proposed target state.
-
-        Returns:
-            True if the transition is allowed by the state machine rules.
-        """
-        return target in _VALID_TRANSITIONS.get(self, set())
-
-    @property
-    def is_idle(self) -> bool:
-        """
-        Check if this state represents inactivity.
-
-        Returns:
-            True if no synchronization is in progress.
-        """
-        return self == SyncState.IDLE
-
-    @property
-    def is_syncing(self) -> bool:
-        """
-        Check if this state represents active synchronization.
-
-        Returns:
-            True if the state involves active block processing.
-        """
-        return self == SyncState.SYNCING
-
-    @property
-    def is_synced(self) -> bool:
-        """
-        Check if this state represents full synchronization.
-
-        Returns:
-            True if the node is caught up with the network.
-        """
-        return self == SyncState.SYNCED
+    """Caught up to the network finalized slot."""
 
     @property
     def accepts_gossip(self) -> bool:
-        """
-        Check if gossip blocks should be processed in this state.
-
-        Returns:
-            True if incoming gossip blocks should be processed.
-        """
+        """Whether incoming gossip blocks should be processed in this state."""
         return self in {SyncState.SYNCING, SyncState.SYNCED}
-
-
-_VALID_TRANSITIONS: dict[SyncState, set[SyncState]] = {
-    SyncState.IDLE: {SyncState.SYNCING},
-    SyncState.SYNCING: {SyncState.SYNCED, SyncState.IDLE},
-    SyncState.SYNCED: {SyncState.SYNCING, SyncState.IDLE},
-}
-"""Valid state transitions for the sync state machine."""

tests/lean_spec/subspecs/sync/test_service.py

@@ -537,6 +537,16 @@ async def test_idle_to_synced_raises_value_error(
         with pytest.raises(ValueError, match="Invalid state transition"):
             await sync_service._transition_to(SyncState.SYNCED)
 
+    async def test_self_transition_raises_value_error(
+        self,
+        sync_service: SyncService,
+    ) -> None:
+        """A transition to the current state is rejected as a no-op move."""
+        assert sync_service.state == SyncState.IDLE
+
+        with pytest.raises(ValueError, match="Invalid state transition"):
+            await sync_service._transition_to(SyncState.IDLE)
+
 
 class TestIdleToCaughtUp:
     """Tests for IDLE-to-SYNCING when already caught up."""

tests/lean_spec/subspecs/sync/test_states.py

@@ -2,8 +2,6 @@
 
 from __future__ import annotations
 
-import pytest
-
 from lean_spec.subspecs.sync.states import SyncState
 
 
@@ -26,110 +24,22 @@ def test_states_are_unique(self) -> None:
         assert len(values) == len(set(values))
 
 
-class TestSyncStateTransitions:
-    """Tests for state transition validation."""
-
-    def test_idle_can_transition_to_syncing(self) -> None:
-        """IDLE can transition to SYNCING."""
-        assert SyncState.IDLE.can_transition_to(SyncState.SYNCING)
-
-    def test_idle_cannot_transition_to_synced(self) -> None:
-        """IDLE cannot transition directly to SYNCED."""
-        assert not SyncState.IDLE.can_transition_to(SyncState.SYNCED)
-
-    def test_syncing_valid_transitions(self) -> None:
-        """SYNCING can transition to SYNCED or IDLE."""
-        assert SyncState.SYNCING.can_transition_to(SyncState.SYNCED)
-        assert SyncState.SYNCING.can_transition_to(SyncState.IDLE)
-
-    def test_synced_valid_transitions(self) -> None:
-        """SYNCED can transition to SYNCING or IDLE."""
-        assert SyncState.SYNCED.can_transition_to(SyncState.SYNCING)
-        assert SyncState.SYNCED.can_transition_to(SyncState.IDLE)
-
-
-class TestSyncStateIsSyncing:
-    """Tests for the is_syncing property."""
-
-    def test_idle_is_not_syncing(self) -> None:
-        """IDLE state is not actively syncing."""
-        assert not SyncState.IDLE.is_syncing
-
-    def test_syncing_is_syncing(self) -> None:
-        """SYNCING state is actively syncing."""
-        assert SyncState.SYNCING.is_syncing
-
-    def test_synced_is_not_syncing(self) -> None:
-        """SYNCED state is not actively syncing."""
-        assert not SyncState.SYNCED.is_syncing
-
-    def test_syncing_states_set(self) -> None:
-        """Exactly one state is a syncing state."""
-        syncing_states = [s for s in SyncState if s.is_syncing]
-        assert len(syncing_states) == 1
-        assert syncing_states[0] == SyncState.SYNCING
-
-
 class TestSyncStateAcceptsGossip:
     """Tests for the accepts_gossip property."""
 
     def test_idle_does_not_accept_gossip(self) -> None:
-        """IDLE state does not accept gossip blocks."""
+        """An idle service ignores incoming gossip blocks."""
         assert not SyncState.IDLE.accepts_gossip
 
     def test_syncing_accepts_gossip(self) -> None:
-        """SYNCING state accepts gossip blocks."""
+        """An actively syncing service processes incoming gossip blocks."""
         assert SyncState.SYNCING.accepts_gossip
 
     def test_synced_accepts_gossip(self) -> None:
-        """SYNCED state accepts gossip blocks."""
+        """A synced service keeps processing gossip blocks for live updates."""
         assert SyncState.SYNCED.accepts_gossip
 
     def test_gossip_accepting_states_set(self) -> None:
-        """Exactly two states accept gossip."""
-        gossip_states = [s for s in SyncState if s.accepts_gossip]
-        assert len(gossip_states) == 2
-        assert set(gossip_states) == {SyncState.SYNCING, SyncState.SYNCED}
-
-
-class TestSyncStateTransitionPaths:
-    """Tests for valid complete transition paths through the state machine."""
-
-    def test_happy_path_to_synced(self) -> None:
-        """Test the happy path: IDLE -> SYNCING -> SYNCED."""
-        current = SyncState.IDLE
-
-        assert current.can_transition_to(SyncState.SYNCING)
-        current = SyncState.SYNCING
-
-        assert current.can_transition_to(SyncState.SYNCED)
-        current = SyncState.SYNCED
-
-        assert current == SyncState.SYNCED
-
-    def test_synced_to_syncing_cycle(self) -> None:
-        """Test SYNCED -> SYNCING for gap handling."""
-        current = SyncState.SYNCED
-
-        assert current.can_transition_to(SyncState.SYNCING)
-        current = SyncState.SYNCING
-
-        assert current.can_transition_to(SyncState.SYNCED)
-        current = SyncState.SYNCED
-
-        assert current == SyncState.SYNCED
-
-
-class TestSyncStateEdgeCases:
-    """Tests for edge cases and invariants."""
-
-    @pytest.mark.parametrize("state", list(SyncState))
-    def test_no_self_transitions(self, state: SyncState) -> None:
-        """No state can transition to itself."""
-        assert not state.can_transition_to(state)
-
-    def test_idle_only_has_one_outgoing_transition(self) -> None:
-        """IDLE has exactly one valid outgoing transition."""
-        valid_targets = [s for s in SyncState if SyncState.IDLE.can_transition_to(s)]
-        assert len(valid_targets) == 1
-        assert valid_targets[0] == SyncState.SYNCING
+        """Exactly the two non-idle states accept gossip."""
+        gossip_states = {s for s in SyncState if s.accepts_gossip}
+        assert gossip_states == {SyncState.SYNCING, SyncState.SYNCED}