refactor: add Interval.from_unix_time and Interval.from_slot classmethods (leanEthereum#544)

Centralizes two repeated conversion patterns into the Interval class:

- from_unix_time: replaces inline (unix - genesis) * 1000 // MS_PER_INTERVAL
- from_slot: replaces inline slot * INTERVALS_PER_SLOT (was duplicated in 6 places)

Adds 14 unit tests covering both classmethods, including a
consistency check that from_slot and from_unix_time agree at
slot boundaries.

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

Author: tcoratger

packages/testing/src/consensus_testing/test_fixtures/fork_choice.py

@@ -12,10 +12,6 @@
 from pydantic import model_validator
 
 from lean_spec.subspecs.chain.clock import Interval
-from lean_spec.subspecs.chain.config import (
-    INTERVALS_PER_SLOT,
-    MILLISECONDS_PER_INTERVAL,
-)
 from lean_spec.subspecs.containers.block import (
     Block,
     BlockBody,
@@ -44,8 +40,6 @@
 )
 from .base import BaseConsensusFixture
 
-DEFAULT_VALIDATOR_ID = ValidatorIndex(0)
-
 
 class ForkChoiceTest(BaseConsensusFixture):
     """
@@ -211,7 +205,7 @@ def make_fixture(self) -> Self:
         store = Store.from_anchor(
             self.anchor_state,
             self.anchor_block,
-            validator_id=DEFAULT_VALIDATOR_ID,
+            validator_id=ValidatorIndex(0),
         )
 
         # Block registry for fork creation
@@ -235,8 +229,9 @@ def make_fixture(self) -> Self:
                         #
                         # TickStep.time is a Unix timestamp in seconds.
                         # Convert to intervals since genesis for the store.
-                        delta_ms = (Uint64(step.time) - store.config.genesis_time) * Uint64(1000)
-                        target_interval = Interval(delta_ms // MILLISECONDS_PER_INTERVAL)
+                        target_interval = Interval.from_unix_time(
+                            Uint64(step.time), store.config.genesis_time
+                        )
                         store, _ = store.on_tick(
                             target_interval, has_proposal=False, is_aggregator=True
                         )
@@ -266,7 +261,7 @@ def make_fixture(self) -> Self:
                         # Store rejects blocks from the future.
                         # This tick includes a block (has proposal).
                         # Always act as aggregator to ensure gossip signatures are aggregated
-                        target_interval = Interval(block.slot * INTERVALS_PER_SLOT)
+                        target_interval = Interval.from_slot(block.slot)
                         store, _ = store.on_tick(
                             target_interval, has_proposal=True, is_aggregator=True
                         )

src/lean_spec/subspecs/chain/clock.py

@@ -18,12 +18,58 @@
 from lean_spec.subspecs.containers.slot import Slot
 from lean_spec.types import Uint64
 
-from .config import MILLISECONDS_PER_INTERVAL, MILLISECONDS_PER_SLOT, SECONDS_PER_SLOT
+from .config import (
+    INTERVALS_PER_SLOT,
+    MILLISECONDS_PER_INTERVAL,
+    MILLISECONDS_PER_SLOT,
+    SECONDS_PER_SLOT,
+)
 
 
 class Interval(Uint64):
     """Interval count since genesis (matches ``Store.time``)."""
 
+    @classmethod
+    def from_unix_time(cls, unix_seconds: Uint64, genesis_time: Uint64) -> Interval:
+        """
+        Convert a Unix timestamp to a total interval count since genesis.
+
+        Useful when external inputs provide absolute timestamps but the
+        store tracks time as intervals (800 ms each, 5 per slot).
+
+        Args:
+            unix_seconds: Absolute Unix timestamp in seconds.
+            genesis_time: Genesis Unix timestamp in seconds.
+
+        Returns:
+            Total intervals elapsed between genesis and the given time.
+        """
+        # Convert the elapsed seconds to milliseconds.
+        #
+        # The store measures time at sub-second granularity (800 ms intervals),
+        # so second-precision input must be scaled up before dividing.
+        delta_ms = (unix_seconds - genesis_time) * Uint64(1000)
+
+        # Truncate to whole intervals.
+        return cls(delta_ms // MILLISECONDS_PER_INTERVAL)
+
+    @classmethod
+    def from_slot(cls, slot: Uint64) -> Interval:
+        """
+        Convert a slot number to the interval at that slot's start.
+
+        Each slot spans a fixed number of intervals.
+        This gives the first interval of the given slot.
+
+        Args:
+            slot: Slot number since genesis.
+
+        Returns:
+            Interval count at the start of the given slot.
+        """
+        # Slot boundaries fall on exact multiples of the interval count.
+        return cls(slot * INTERVALS_PER_SLOT)
+
 
 @dataclass(frozen=True, slots=True)
 class SlotClock:

src/lean_spec/subspecs/forkchoice/store.py

@@ -221,7 +221,7 @@ def from_anchor(
         # The state internally might have zero-hash checkpoints (if genesis),
         # but the Store must treat the anchor block as the justified/finalized point.
         return cls(
-            time=Interval(anchor_slot * INTERVALS_PER_SLOT),
+            time=Interval.from_slot(anchor_slot),
             config=state.config,
             head=anchor_root,
             safe_target=anchor_root,
@@ -1191,7 +1191,7 @@ def get_proposal_head(self, slot: Slot) -> tuple["Store", Bytes32]:
             Tuple of (new Store with updated time, head root for building).
         """
         # Advance time to this slot's first interval
-        target_interval = Interval(slot * INTERVALS_PER_SLOT)
+        target_interval = Interval.from_slot(slot)
         store, _ = self.on_tick(target_interval, True)
 
         # Process any pending attestations before proposal

src/lean_spec/subspecs/node/node.py

@@ -22,10 +22,7 @@
 from lean_spec.subspecs.api import ApiServer, ApiServerConfig
 from lean_spec.subspecs.chain import SlotClock
 from lean_spec.subspecs.chain.clock import Interval
-from lean_spec.subspecs.chain.config import (
-    ATTESTATION_COMMITTEE_COUNT,
-    INTERVALS_PER_SLOT,
-)
+from lean_spec.subspecs.chain.config import ATTESTATION_COMMITTEE_COUNT
 from lean_spec.subspecs.chain.service import ChainService
 from lean_spec.subspecs.containers import Block, BlockBody, SignedBlock, State
 from lean_spec.subspecs.containers.attestation import SignedAttestation
@@ -393,7 +390,7 @@ def _try_load_store_from_database(
         # Instead, derive time from wall clock, floored by the block's slot.
         clock = SlotClock(genesis_time=genesis_time or _ZERO_TIME, time_fn=time_fn)
         wall_clock_intervals = clock.total_intervals()
-        block_intervals = Interval(head_block.slot * INTERVALS_PER_SLOT)
+        block_intervals = Interval.from_slot(head_block.slot)
         store_time = max(wall_clock_intervals, block_intervals)
 
         # Reconstruct minimal store from persisted data.

tests/lean_spec/helpers/builders.py

@@ -12,7 +12,6 @@
 from consensus_testing.keys import XmssKeyManager
 
 from lean_spec.subspecs.chain.clock import Interval, SlotClock
-from lean_spec.subspecs.chain.config import INTERVALS_PER_SLOT
 from lean_spec.subspecs.containers import (
     AttestationData,
     Block,
@@ -464,7 +463,7 @@ def make_signed_block_from_store(
         ),
     )
 
-    target_interval = Interval(block.slot * INTERVALS_PER_SLOT)
+    target_interval = Interval.from_slot(block.slot)
     advanced_store, _ = store.on_tick(target_interval, has_proposal=True)
 
     return advanced_store, signed_block

tests/lean_spec/subspecs/chain/test_clock.py

@@ -1,15 +1,119 @@
 """Tests for the SlotClock time-to-slot converter."""
 
+from __future__ import annotations
+
 import pytest
 
 from lean_spec.subspecs.chain import Interval, SlotClock
 from lean_spec.subspecs.chain.config import (
+    INTERVALS_PER_SLOT,
     MILLISECONDS_PER_INTERVAL,
+    MILLISECONDS_PER_SLOT,
     SECONDS_PER_SLOT,
 )
 from lean_spec.subspecs.containers import Slot
 from lean_spec.types import Uint64
 
+GENESIS_TIME = Uint64(1_700_000_000)
+
+
+class TestIntervalFromUnixTime:
+    """Tests for Interval.from_unix_time()."""
+
+    def test_at_genesis(self) -> None:
+        """Returns interval 0 when unix_seconds equals genesis_time."""
+        assert Interval.from_unix_time(GENESIS_TIME, GENESIS_TIME) == Interval(0)
+
+    def test_one_second_after_genesis(self) -> None:
+        """One second equals 1000ms, yielding 1000 // 800 = 1 interval."""
+        result = Interval.from_unix_time(GENESIS_TIME + Uint64(1), GENESIS_TIME)
+        assert result == Interval(1)
+
+    def test_one_slot_after_genesis(self) -> None:
+        """One full slot (4s = 4000ms) yields 4000 // 800 = 5 intervals."""
+        result = Interval.from_unix_time(GENESIS_TIME + SECONDS_PER_SLOT, GENESIS_TIME)
+        expected = Interval(int(MILLISECONDS_PER_SLOT // MILLISECONDS_PER_INTERVAL))
+        assert result == expected
+
+    def test_sub_interval_rounds_down(self) -> None:
+        """Partial intervals are truncated by integer division.
+
+        At 0 full seconds past genesis the delta_ms is 0, so the result is 0.
+        The method only accepts whole-second Uint64 values, so sub-interval
+        precision only surfaces when 1000ms is not a multiple of the interval.
+        Here we verify the floor behaviour across the first few seconds.
+        """
+        # 0s -> 0ms -> 0 intervals
+        assert Interval.from_unix_time(GENESIS_TIME, GENESIS_TIME) == Interval(0)
+        # 1s -> 1000ms -> 1000 // 800 = 1 (remainder 200ms truncated)
+        assert Interval.from_unix_time(GENESIS_TIME + Uint64(1), GENESIS_TIME) == Interval(1)
+        # 2s -> 2000ms -> 2000 // 800 = 2 (remainder 400ms truncated)
+        assert Interval.from_unix_time(GENESIS_TIME + Uint64(2), GENESIS_TIME) == Interval(2)
+        # 3s -> 3000ms -> 3000 // 800 = 3 (remainder 600ms truncated)
+        assert Interval.from_unix_time(GENESIS_TIME + Uint64(3), GENESIS_TIME) == Interval(3)
+
+    def test_multiple_slots(self) -> None:
+        """Ten slots (40s = 40000ms) yields 40000 // 800 = 50 intervals."""
+        ten_slots = Uint64(10) * SECONDS_PER_SLOT
+        result = Interval.from_unix_time(GENESIS_TIME + ten_slots, GENESIS_TIME)
+        expected_intervals = (ten_slots * Uint64(1000)) // MILLISECONDS_PER_INTERVAL
+        assert result == Interval(expected_intervals)
+
+    def test_return_type_is_interval(self) -> None:
+        """Return value is an Interval instance, not a plain Uint64."""
+        result = Interval.from_unix_time(GENESIS_TIME + Uint64(5), GENESIS_TIME)
+        assert isinstance(result, Interval)
+
+    def test_large_time_delta(self) -> None:
+        """Works correctly with a large time delta (one day = 86400s)."""
+        one_day = Uint64(86400)
+        result = Interval.from_unix_time(GENESIS_TIME + one_day, GENESIS_TIME)
+        expected = Interval((one_day * Uint64(1000)) // MILLISECONDS_PER_INTERVAL)
+        assert result == expected
+
+    def test_genesis_time_zero(self) -> None:
+        """Works when genesis_time is zero."""
+        result = Interval.from_unix_time(Uint64(4), Uint64(0))
+        # 4s = 4000ms -> 4000 // 800 = 5
+        assert result == Interval(5)
+
+
+class TestIntervalFromSlot:
+    """Tests for Interval.from_slot()."""
+
+    def test_slot_zero(self) -> None:
+        """Slot 0 maps to interval 0."""
+        assert Interval.from_slot(Uint64(0)) == Interval(0)
+
+    def test_slot_one(self) -> None:
+        """Slot 1 maps to interval equal to INTERVALS_PER_SLOT."""
+        assert Interval.from_slot(Uint64(1)) == Interval(INTERVALS_PER_SLOT)
+
+    def test_slot_three(self) -> None:
+        """Slot 3 maps to interval 3 * INTERVALS_PER_SLOT."""
+        assert Interval.from_slot(Uint64(3)) == Interval(Uint64(3) * INTERVALS_PER_SLOT)
+
+    def test_multiple_slots(self) -> None:
+        """Each slot N maps to interval N * INTERVALS_PER_SLOT."""
+        for n in range(10):
+            slot = Uint64(n)
+            assert Interval.from_slot(slot) == Interval(slot * INTERVALS_PER_SLOT)
+
+    def test_return_type_is_interval(self) -> None:
+        """Return value is an Interval instance, not a plain Uint64."""
+        result = Interval.from_slot(Uint64(2))
+        assert isinstance(result, Interval)
+
+    def test_consistent_with_from_unix_time(self) -> None:
+        """from_slot(N) equals from_unix_time at genesis + N * SECONDS_PER_SLOT."""
+        for n in range(5):
+            slot = Uint64(n)
+            from_slot_result = Interval.from_slot(slot)
+            from_unix_result = Interval.from_unix_time(
+                GENESIS_TIME + slot * SECONDS_PER_SLOT, GENESIS_TIME
+            )
+            assert from_slot_result == from_unix_result
+
 
 class TestCurrentSlot:
     """Tests for current_slot()."""

tests/lean_spec/subspecs/forkchoice/test_attestation_target.py

@@ -6,10 +6,7 @@
 from consensus_testing.keys import XmssKeyManager
 
 from lean_spec.subspecs.chain.clock import Interval
-from lean_spec.subspecs.chain.config import (
-    INTERVALS_PER_SLOT,
-    JUSTIFICATION_LOOKBACK_SLOTS,
-)
+from lean_spec.subspecs.chain.config import JUSTIFICATION_LOOKBACK_SLOTS
 from lean_spec.subspecs.containers import (
     Attestation,
     AttestationData,
@@ -567,7 +564,7 @@ def test_attestation_target_after_on_block(
 
         # Process block via on_block on a fresh consumer store
         consumer_store = observer_store
-        target_interval = Interval(block.slot * INTERVALS_PER_SLOT)
+        target_interval = Interval.from_slot(block.slot)
         consumer_store, _ = consumer_store.on_tick(target_interval, has_proposal=True)
         consumer_store = consumer_store.on_block(signed_block)