cryptarchia: Update epoch stabilization schedule to 334 (from 433) (#79)

* feat(cryptarchia/epoch-schedule): switch to 334 schedule (from 433)

* factor out common test config building code

* feat(cryptarchia): test_leader uses common test config object

* cryptarchia: update test_ledger_state_is_properly_updated_on_reorg

* cryptarchia: update test_epoch_transition test

* move to .tip() in tests instead of tip_id()

* cryptarchia: wrap long comments

* cryptarchia: move mk_block to test_common

* cryptarchia: move mk_genesis_state to test_common

* cryptarchia: refactor fork test to use mk_chain

* cryptarchia: fork choice rules tests use mk_chain helper

* cryptarchia: rename fork choice test suite to TestForkChoice

* cryptarchia: config.s is always 3k/f or 3*base_period_length

* cryptarchia: hardcode epoch schedule in specification

* un-hard code epoch sched. params + provide a v0.0.1 spec for params

cryptarchia/cryptarchia.py

@@ -6,7 +6,7 @@ from itertools import chain
 import functools
 
 # Please note this is still a work in progress
-from dataclasses import dataclass, field
+from dataclasses import dataclass, field, replace
 
 Id: TypeAlias = bytes
 
@@ -27,8 +27,9 @@ class TimeConfig:
 
 @dataclass
 class Config:
-    k: int
+    k: int  # The depth of a block before it is considered immutable.
     active_slot_coeff: float  # 'f', the rate of occupied slots
+
     # The stake distribution is always taken at the beginning of the previous epoch.
     # This parameters controls how many slots to wait for it to be stabilized
     # The value is computed as epoch_stake_distribution_stabilization * int(floor(k / f))
@@ -39,8 +40,23 @@ class Config:
     # This parameter controls how many slots we wait for the nonce snapshot to be considered
     # stabilized
     epoch_period_nonce_stabilization: int
+
     time: TimeConfig
 
+    @staticmethod
+    def cryptarchia_v0_0_1() -> "Config":
+        return Config(
+            k=2160,
+            active_slot_coeff=0.05,
+            epoch_stake_distribution_stabilization=3,
+            epoch_period_nonce_buffer=3,
+            epoch_period_nonce_stabilization=4,
+            time=TimeConfig(
+                slot_duration=1,
+                chain_start_time=0,
+            ),
+        )
+
     @property
     def base_period_length(self) -> int:
         return int(floor(self.k / self.active_slot_coeff))
@@ -55,7 +71,14 @@ class Config:
 
     @property
     def s(self):
-        return self.base_period_length * self.epoch_period_nonce_stabilization
+        """
+        The Security Paramater. This paramter controls how many slots one must wait before we
+        have high confidence that k blocks have been produced.
+        """
+        return self.base_period_length * 3
+
+    def replace(self, **kwarg) -> "Config":
+        return replace(self, **kwarg)
 
 
 # An absolute unique indentifier of a slot, counting incrementally from 0
@@ -448,6 +471,9 @@ class Follower:
         else:
             return self.genesis_state.block
 
+    def tip_state(self) -> LedgerState:
+        return self.ledger_state[self.tip_id()]
+
     def state_at_slot_beginning(self, chain: Chain, slot: Slot) -> LedgerState:
         for block in reversed(chain.blocks):
             if block.slot < slot:

cryptarchia/test_common.py

@@ -0,0 +1,54 @@
+from .cryptarchia import (
+    Config,
+    TimeConfig,
+    Id,
+    Slot,
+    Coin,
+    BlockHeader,
+    LedgerState,
+    MockLeaderProof,
+)
+
+
+def mk_config() -> Config:
+    return Config.cryptarchia_v0_0_1().replace(
+        k=1,
+        active_slot_coeff=1.0,
+    )
+
+
+def mk_genesis_state(initial_stake_distribution: list[Coin]) -> LedgerState:
+    return LedgerState(
+        block=bytes(32),
+        nonce=bytes(32),
+        total_stake=sum(c.value for c in initial_stake_distribution),
+        commitments_spend={c.commitment() for c in initial_stake_distribution},
+        commitments_lead={c.commitment() for c in initial_stake_distribution},
+        nullifiers=set(),
+    )
+
+
+def mk_block(
+    parent: Id, slot: int, coin: Coin, content=bytes(32), orphaned_proofs=[]
+) -> BlockHeader:
+    assert len(parent) == 32
+    from hashlib import sha256
+
+    return BlockHeader(
+        slot=Slot(slot),
+        parent=parent,
+        content_size=len(content),
+        content_id=sha256(content).digest(),
+        leader_proof=MockLeaderProof.new(coin, Slot(slot), parent=parent),
+        orphaned_proofs=orphaned_proofs,
+    )
+
+
+def mk_chain(parent, coin: Coin, slots: list[int]) -> tuple[list[BlockHeader], Coin]:
+    chain = []
+    for s in slots:
+        block = mk_block(parent=parent, slot=s, coin=coin)
+        chain.append(block)
+        parent = block.id()
+        coin = coin.evolve()
+    return chain, coin

cryptarchia/test_fork_choice.py

@@ -14,89 +14,63 @@ from cryptarchia.cryptarchia import (
     Coin,
 )
 
-
+from .test_common import mk_chain
-def make_block(parent_id: Id, slot: Slot, content: bytes) -> BlockHeader:
-    assert len(parent_id) == 32
-    content_id = hashlib.sha256(content).digest()
-    return BlockHeader(
-        parent=parent_id,
-        content_size=1,
-        slot=slot,
-        content_id=content_id,
-        leader_proof=MockLeaderProof.new(
-            Coin(sk=0, value=10), slot=slot, parent=parent_id
-        ),
-    )
 
 
-class TestLeader(TestCase):
+class TestForkChoice(TestCase):
     def test_fork_choice_long_sparse_chain(self):
         # The longest chain is not dense after the fork
-        common = [make_block(bytes(32), Slot(i), bytes(i)) for i in range(1, 50)]
+        short_coin, long_coin = Coin(sk=0, value=100), Coin(sk=1, value=100)
-        long_chain = deepcopy(common)
+        common, long_coin = mk_chain(parent=bytes(32), coin=long_coin, slots=range(50))
-        short_chain = deepcopy(common)
 
-        for slot in range(50, 100):
+        long_chain_sparse_ext, long_coin = mk_chain(
-            # make arbitrary ids for the different chain so that the blocks appear to be different
+            parent=common[-1].id(), coin=long_coin, slots=range(50, 100, 2)
-            long_content = f"{slot}-long".encode()
+        )
-            short_content = f"{slot}-short".encode()
+
-            if slot % 2 == 0:
+        short_chain_dense_ext, _ = mk_chain(
-                long_chain.append(make_block(bytes(32), Slot(slot), long_content))
+            parent=common[-1].id(), coin=short_coin, slots=range(50, 100)
-            short_chain.append(make_block(bytes(32), Slot(slot), short_content))
+        )
-        # add more blocks to the long chain
+
-        for slot in range(100, 200):
+        # add more blocks to the long chain to ensure the long chain is indeed longer
-            long_content = f"{slot}-long".encode()
+        long_chain_further_ext, _ = mk_chain(
-            long_chain.append(make_block(bytes(32), Slot(slot), long_content))
+            parent=long_chain_sparse_ext[-1].id(), coin=long_coin, slots=range(100, 126)
+        )
+
+        long_chain = deepcopy(common) + long_chain_sparse_ext + long_chain_further_ext
+        short_chain = deepcopy(common) + short_chain_dense_ext
         assert len(long_chain) > len(short_chain)
+
         # by setting a low k we trigger the density choice rule
         k = 1
         s = 50
+
         short_chain = Chain(short_chain, genesis=bytes(32))
         long_chain = Chain(long_chain, genesis=bytes(32))
-        assert (
+        assert maxvalid_bg(short_chain, [long_chain], k, s) == short_chain
-            maxvalid_bg(
-                short_chain,
-                [long_chain],
-                k,
-                s,
-            )
-            == short_chain
-        )
 
         # However, if we set k to the fork length, it will be accepted
         k = long_chain.length()
-        assert (
+        assert maxvalid_bg(short_chain, [long_chain], k, s) == long_chain
-            maxvalid_bg(
-                short_chain,
-                [long_chain],
-                k,
-                s,
-            )
-            == long_chain
-        )
 
     def test_fork_choice_long_dense_chain(self):
         # The longest chain is also the densest after the fork
-        common = [make_block(bytes(32), Slot(i), bytes(i)) for i in range(1, 50)]
+        short_coin, long_coin = Coin(sk=0, value=100), Coin(sk=1, value=100)
-        long_chain = deepcopy(common)
+        common, long_coin = mk_chain(
-        short_chain = deepcopy(common)
+            parent=bytes(32), coin=long_coin, slots=range(1, 50)
-        for slot in range(50, 100):
+        )
-            # make arbitrary ids for the different chain so that the blocks appear to be different
+
-            long_content = f"{slot}-long".encode()
+        long_chain_dense_ext, _ = mk_chain(
-            short_content = f"{slot}-short".encode()
+            parent=common[-1].id(), coin=long_coin, slots=range(50, 100)
-            long_chain.append(make_block(bytes(32), Slot(slot), long_content))
+        )
-            if slot % 2 == 0:
+        short_chain_sparse_ext, _ = mk_chain(
-                short_chain.append(make_block(bytes(32), Slot(slot), short_content))
+            parent=common[-1].id(), coin=short_coin, slots=range(50, 100, 2)
+        )
+
+        long_chain = deepcopy(common) + long_chain_dense_ext
+        short_chain = deepcopy(common) + short_chain_sparse_ext
+
         k = 1
         s = 50
         short_chain = Chain(short_chain, genesis=bytes(32))
         long_chain = Chain(long_chain, genesis=bytes(32))
-        assert (
+        assert maxvalid_bg(short_chain, [long_chain], k, s) == long_chain
-            maxvalid_bg(
-                short_chain,
-                [long_chain],
-                k,
-                s,
-            )
-            == long_chain
-        )

cryptarchia/test_leader.py

@@ -12,6 +12,7 @@ from .cryptarchia import (
     TimeConfig,
     Slot,
 )
+from .test_common import mk_config
 
 
 class TestLeader(TestCase):
@@ -24,15 +25,10 @@ class TestLeader(TestCase):
         )
 
         f = 0.05
-        config = Config(
+        l = Leader(
-            k=10,
+            config=mk_config().replace(active_slot_coeff=f),
-            active_slot_coeff=f,
+            coin=Coin(sk=0, value=10),
-            epoch_stake_distribution_stabilization=4,
-            epoch_period_nonce_buffer=3,
-            epoch_period_nonce_stabilization=3,
-            time=TimeConfig(slot_duration=1, chain_start_time=0),
         )
-        l = Leader(config=config, coin=Coin(sk=0, value=10))
 
         # We'll use the Margin of Error equation to decide how many samples we need.
         # https://en.wikipedia.org/wiki/Margin_of_error
@@ -42,7 +38,8 @@ class TestLeader(TestCase):
         Z = 3  # we want 3 std from the mean to be within the margin of error
         N = int((Z * std / margin_of_error) ** 2)
 
-        # After N slots, the measured leader rate should be within the interval `p +- margin_of_error` with high probabiltiy
+        # After N slots, the measured leader rate should be within the
+        # interval `p +- margin_of_error` with high probabiltiy
         leader_rate = (
             sum(
                 l.try_prove_slot_leader(epoch, Slot(slot), bytes(32)) is not None

cryptarchia/test_ledger_state_update.py

@@ -14,42 +14,7 @@ from .cryptarchia import (
     Id,
 )
 
-
+from .test_common import mk_config, mk_block, mk_genesis_state
-def mk_genesis_state(initial_stake_distribution: list[Coin]) -> LedgerState:
-    return LedgerState(
-        block=bytes(32),
-        nonce=bytes(32),
-        total_stake=sum(c.value for c in initial_stake_distribution),
-        commitments_spend={c.commitment() for c in initial_stake_distribution},
-        commitments_lead={c.commitment() for c in initial_stake_distribution},
-        nullifiers=set(),
-    )
-
-
-def mk_block(
-    parent: Id, slot: int, coin: Coin, content=bytes(32), orphaned_proofs=[]
-) -> BlockHeader:
-    from hashlib import sha256
-
-    return BlockHeader(
-        slot=Slot(slot),
-        parent=parent,
-        content_size=len(content),
-        content_id=sha256(content).digest(),
-        leader_proof=MockLeaderProof.new(coin, Slot(slot), parent=parent),
-        orphaned_proofs=orphaned_proofs,
-    )
-
-
-def config() -> Config:
-    return Config(
-        k=10,
-        active_slot_coeff=0.05,
-        epoch_stake_distribution_stabilization=4,
-        epoch_period_nonce_buffer=3,
-        epoch_period_nonce_stabilization=3,
-        time=TimeConfig(slot_duration=1, chain_start_time=0),
-    )
 
 
 class TestLedgerStateUpdate(TestCase):
@@ -57,27 +22,24 @@ class TestLedgerStateUpdate(TestCase):
         leader_coin = Coin(sk=0, value=100)
         genesis = mk_genesis_state([leader_coin])
 
-        follower = Follower(genesis, config())
+        follower = Follower(genesis, mk_config())
 
         block = mk_block(slot=0, parent=genesis.block, coin=leader_coin)
         follower.on_block(block)
 
         # Follower should have accepted the block
         assert follower.local_chain.length() == 1
-        assert follower.local_chain.tip() == block
+        assert follower.tip() == block
 
         # Follower should have updated their ledger state to mark the leader coin as spent
-        assert (
+        assert follower.tip_state().verify_unspent(leader_coin.nullifier()) == False
-            follower.ledger_state[block.id()].verify_unspent(leader_coin.nullifier())
-            == False
-        )
 
-        reuse_coin_block = mk_block(slot=1, parent=block.id, coin=leader_coin)
+        reuse_coin_block = mk_block(slot=1, parent=block.id(), coin=leader_coin)
         follower.on_block(block)
 
         # Follower should *not* have accepted the block
         assert follower.local_chain.length() == 1
-        assert follower.local_chain.tip() == block
+        assert follower.tip() == block
 
     def test_ledger_state_is_properly_updated_on_reorg(self):
         coin_1 = Coin(sk=0, value=100)
@@ -86,7 +48,7 @@ class TestLedgerStateUpdate(TestCase):
 
         genesis = mk_genesis_state([coin_1, coin_2, coin_3])
 
-        follower = Follower(genesis, config())
+        follower = Follower(genesis, mk_config())
 
         # 1) coin_1 & coin_2 both concurrently win slot 0
 
@@ -96,15 +58,13 @@ class TestLedgerStateUpdate(TestCase):
         # 2) follower sees block 1 first
 
         follower.on_block(block_1)
-        assert follower.tip_id() == block_1.id()
+        assert follower.tip() == block_1
-        assert not follower.ledger_state[block_1.id()].verify_unspent(
+        assert not follower.tip_state().verify_unspent(coin_1.nullifier())
-            coin_1.nullifier()
-        )
 
         # 3) then sees block 2, but sticks with block_1 as the tip
 
         follower.on_block(block_2)
-        assert follower.tip_id() == block_1.id()
+        assert follower.tip() == block_1
         assert len(follower.forks) == 1, f"{len(follower.forks)}"
 
         # 4) then coin_3 wins slot 1 and chooses to extend from block_2
@@ -112,40 +72,44 @@ class TestLedgerStateUpdate(TestCase):
         block_3 = mk_block(parent=block_2.id(), slot=1, coin=coin_3)
         follower.on_block(block_3)
         # the follower should have switched over to the block_2 fork
-        assert follower.tip_id() == block_3.id()
+        assert follower.tip() == block_3
 
         # and the original coin_1 should now be removed from the spent pool
-        assert follower.ledger_state[block_3.id()].verify_unspent(coin_1.nullifier())
+        assert follower.tip_state().verify_unspent(coin_1.nullifier())
 
     def test_epoch_transition(self):
         leader_coins = [Coin(sk=i, value=100) for i in range(4)]
         genesis = mk_genesis_state(leader_coins)
-
+        config = mk_config()
-        # An epoch will be 10 slots long, with stake distribution snapshot taken at the start of the epoch
-        # and nonce snapshot before slot 7
-        config = Config(
-            k=1,
-            active_slot_coeff=1,
-            epoch_stake_distribution_stabilization=4,
-            epoch_period_nonce_buffer=3,
-            epoch_period_nonce_stabilization=3,
-            time=TimeConfig(slot_duration=1, chain_start_time=0),
-        )
 
         follower = Follower(genesis, config)
+
+        # We assume an epoch length of 10 slots in this test.
+        assert config.epoch_length == 10, f"epoch len: {config.epoch_length}"
+
+        # ---- EPOCH 0 ----
+
         block_1 = mk_block(slot=0, parent=genesis.block, coin=leader_coins[0])
         follower.on_block(block_1)
         assert follower.tip() == block_1
-        assert follower.tip().slot.epoch(follower.config).epoch == 0
+        assert follower.tip().slot.epoch(config).epoch == 0
+
         block_2 = mk_block(slot=9, parent=block_1.id(), coin=leader_coins[1])
         follower.on_block(block_2)
         assert follower.tip() == block_2
-        assert follower.tip().slot.epoch(follower.config).epoch == 0
+        assert follower.tip().slot.epoch(config).epoch == 0
+
+        # ---- EPOCH 1 ----
+
         block_3 = mk_block(slot=10, parent=block_2.id(), coin=leader_coins[2])
         follower.on_block(block_3)
+        assert follower.tip() == block_3
+        assert follower.tip().slot.epoch(config).epoch == 1
 
-        # when trying to propose a block for epoch 2, the stake distribution snapshot should be taken at the end
+        # ---- EPOCH 2 ----
-        # of epoch 1, i.e. slot 9
+
+        # when trying to propose a block for epoch 2, the stake distribution snapshot should be taken
+        # at the end of epoch 0, i.e. slot 9
         # To ensure this is the case, we add a new coin just to the state associated with that slot,
         # so that the new block can be accepted only if that is the snapshot used
         # first, verify that if we don't change the state, the block is not accepted
@@ -158,56 +122,39 @@ class TestLedgerStateUpdate(TestCase):
         )
         follower.on_block(block_4)
         assert follower.tip() == block_4
-        assert follower.tip().slot.epoch(follower.config).epoch == 2
+        assert follower.tip().slot.epoch(config).epoch == 2
 
     def test_evolved_coin_is_eligible_for_leadership(self):
         coin = Coin(sk=0, value=100)
 
         genesis = mk_genesis_state([coin])
 
-        config = Config(
+        follower = Follower(genesis, mk_config())
-            k=1,
-            active_slot_coeff=1,
-            epoch_stake_distribution_stabilization=4,
-            epoch_period_nonce_buffer=3,
-            epoch_period_nonce_stabilization=3,
-            time=TimeConfig(slot_duration=1, chain_start_time=0),
-        )
-
-        follower = Follower(genesis, config)
 
         # coin wins the first slot
         block_1 = mk_block(slot=0, parent=genesis.block, coin=coin)
         follower.on_block(block_1)
-        assert follower.tip_id() == block_1.id()
+        assert follower.tip() == block_1
 
         # coin can't be reused to win following slots:
         block_2_reuse = mk_block(slot=1, parent=block_1.id(), coin=coin)
         follower.on_block(block_2_reuse)
-        assert follower.tip_id() == block_1.id()
+        assert follower.tip() == block_1
 
         # but the evolved coin is eligible
         block_2_evolve = mk_block(slot=1, parent=block_1.id(), coin=coin.evolve())
         follower.on_block(block_2_evolve)
-        assert follower.tip_id() == block_2_evolve.id()
+        assert follower.tip() == block_2_evolve
 
     def test_new_coins_becoming_eligible_after_stake_distribution_stabilizes(self):
+        config = mk_config()
         coin = Coin(sk=0, value=100)
         genesis = mk_genesis_state([coin])
-
-        # An epoch will be 10 slots long, with stake distribution snapshot taken at the start of the epoch
-        # and nonce snapshot before slot 7
-        config = Config(
-            k=1,
-            active_slot_coeff=1,
-            epoch_stake_distribution_stabilization=4,
-            epoch_period_nonce_buffer=3,
-            epoch_period_nonce_stabilization=3,
-            time=TimeConfig(slot_duration=1, chain_start_time=0),
-        )
-
         follower = Follower(genesis, config)
 
+        # We assume an epoch length of 10 slots in this test.
+        assert config.epoch_length == 10
+
         # ---- EPOCH 0 ----
 
         block_0_0 = mk_block(slot=0, parent=genesis.block, coin=coin)
@@ -221,13 +168,11 @@ class TestLedgerStateUpdate(TestCase):
         )
 
         # the new coin is not yet eligible for elections
-
         block_0_1_attempt = mk_block(slot=1, parent=block_0_0.id(), coin=coin_new)
         follower.on_block(block_0_1_attempt)
         assert follower.tip() == block_0_0
 
         # whereas the evolved coin from genesis can be spent immediately
-
         block_0_1 = mk_block(slot=1, parent=block_0_0.id(), coin=coin.evolve())
         follower.on_block(block_0_1)
         assert follower.tip() == block_0_1
@@ -245,7 +190,11 @@ class TestLedgerStateUpdate(TestCase):
 
         # The coin is finally eligible 2 epochs after it was first minted
 
-        block_2_0 = mk_block(slot=20, parent=block_0_1.id(), coin=coin_new)
+        block_2_0 = mk_block(
+            slot=20,
+            parent=block_0_1.id(),
+            coin=coin_new,
+        )
         follower.on_block(block_2_0)
         assert follower.tip() == block_2_0
 
@@ -259,20 +208,7 @@ class TestLedgerStateUpdate(TestCase):
         coin = Coin(sk=0, value=100)
         genesis = mk_genesis_state([coin])
 
-        # An epoch will be 10 slots long, with stake distribution snapshot taken at the start of the epoch
+        follower = Follower(genesis, mk_config())
-        # and nonce snapshot before slot 7
-        config = Config(
-            k=1,
-            active_slot_coeff=1,
-            epoch_stake_distribution_stabilization=4,
-            epoch_period_nonce_buffer=3,
-            epoch_period_nonce_stabilization=3,
-            time=TimeConfig(slot_duration=1, chain_start_time=0),
-        )
-
-        follower = Follower(genesis, config)
-
-        # ---- EPOCH 0 ----
 
         block_0_0 = mk_block(slot=0, parent=genesis.block, coin=coin)
         follower.on_block(block_0_0)
@@ -284,7 +220,9 @@ class TestLedgerStateUpdate(TestCase):
         follower.on_block(block_0_1)
         # the coin evolved twice should not be accepted as it is not in the lead commitments
         assert follower.tip() == block_0_0
-        # an orphaned proof with an evolved coin for the same slot as the original coin should not be accepted as the evolved coin is not in the lead commitments at slot 0
+
+        # an orphaned proof with an evolved coin for the same slot as the original coin
+        # should not be accepted as the evolved coin is not in the lead commitments at slot 0
         block_0_1 = mk_block(
             slot=1,
             parent=block_0_0.id(),
@@ -293,7 +231,9 @@ class TestLedgerStateUpdate(TestCase):
         )
         follower.on_block(block_0_1)
         assert follower.tip() == block_0_0
-        # the coin evolved twice should be accepted as the evolved coin is in the lead commitments at slot 1 and processed before that
+
+        # the coin evolved twice should be accepted as the evolved coin is in the lead commitments
+        # at slot 1 and processed before that
         block_0_2 = mk_block(
             slot=2,
             parent=block_0_0.id(),