logos-blockchain/logos-blockchain-specs
1
0
Fork 0
mirror of https://github.com/logos-blockchain/logos-blockchain-specs.git synced 2026-01-05 22:53:11 +00:00
Code Issues Packages Projects Releases Wiki Activity

Cryptarchia: rework specification (#116)

Browse Source 
* cryptarchia/ghost: prep for move to weight based fork choice

* cryptarchia/ghost: remove common_prefix_len helper

* cryptarchia/ghost: common_prefix_depth returns depth of both chains

* cryptarchia/ghost: fix chain density calculation

* cryptarchia/ghost: maxvalid_bg uses block ids rather than chains

* cryptarchia/ghost: unimported_orphans returns orphans w.r.t. to tip

* cryptarchia/ghost: remove redundant check

* cryptarchia/ghost: rewrite unimported_orphan w/ common_prefix_depths

* cryptarchia/ghost: validate_header w.r.t. block parent

* cryptachia/ghost: rewrite on_block to remove dependency on Chain

* cryptarchia/ghost: remove Chain abstraction

* cryptarchia/ghost: remove local / fork naming in common_prefix_depth

* cryptarchia/ghost: rewrite common_prefix_depth in terms of iter_chain

* cryptarchia/ghost: impl GHOST fork choice rule

* cryptarchia/ghost: integrate GHOST with maxvalid fork choice

* cryptarchia: remove unused imports

* cryptarchia: cleanup

* cryptarchia: cleanup

* cryptarchia: remove height from ledger state

* cryptachia/ghost: update fork choice rule comments

* cryptarchia: switch back to longest chain

* cryptarchia: update tests

* cryptarchia: remove debug log
This commit is contained in:
davidrusu 2025-02-25 16:54:54 +04:00 committed by GitHub
parent 3f3427ee9f
commit 5c64a0bd11
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 359 additions and 309 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

306
cryptarchia/cryptarchia.py
View File

@ -1,11 +1,12 @@
from typing import TypeAlias, List, Optional from typing import TypeAlias, List, Dict
from hashlib import sha256, blake2b from hashlib import sha256, blake2b
from math import floor from math import floor
from copy import deepcopy from copy import deepcopy
from itertools import chain import itertools
import functools import functools
from dataclasses import dataclass, field, replace from dataclasses import dataclass, field, replace
import logging import logging
from collections import defaultdict
import numpy as np import numpy as np
@ -248,36 +249,13 @@ class BlockHeader:
return h.digest() return h.digest()
@dataclass
class Chain:
blocks: List[BlockHeader]
genesis: Id
def tip_id(self) -> Id:
if len(self.blocks) == 0:
return self.genesis
return self.tip().id()
def tip(self) -> BlockHeader:
return self.blocks[-1]
def length(self) -> int:
return len(self.blocks)
def block_position(self, block: Id) -> Optional[int]:
for i, b in enumerate(self.blocks):
if b.id() == block:
return i
return None
@dataclass @dataclass
class LedgerState: class LedgerState:
""" """
A snapshot of the ledger state up to some block A snapshot of the ledger state up to some block
""" """
block: Id = None block: BlockHeader
# This nonce is used to derive the seed for the slot leader lottery. # This nonce is used to derive the seed for the slot leader lottery.
# It's updated at every block by hashing the previous nonce with the # It's updated at every block by hashing the previous nonce with the
@ -324,7 +302,7 @@ class LedgerState:
return nullifier not in self.nullifiers return nullifier not in self.nullifiers
def apply(self, block: BlockHeader): def apply(self, block: BlockHeader):
assert block.parent == self.block assert block.parent == self.block.id()
h = blake2b(digest_size=32) h = blake2b(digest_size=32)
h.update("epoch-nonce".encode(encoding="utf-8")) h.update("epoch-nonce".encode(encoding="utf-8"))
@ -333,8 +311,8 @@ class LedgerState:
h.update(block.slot.encode()) h.update(block.slot.encode())
self.nonce = h.digest() self.nonce = h.digest()
self.block = block.id() self.block = block
for proof in chain(block.orphaned_proofs, [block]): for proof in itertools.chain(block.orphaned_proofs, [block]):
self.apply_leader_proof(proof.leader_proof) self.apply_leader_proof(proof.leader_proof)
def apply_leader_proof(self, proof: MockLeaderProof): def apply_leader_proof(self, proof: MockLeaderProof):
@ -385,19 +363,25 @@ class Follower:
def __init__(self, genesis_state: LedgerState, config: Config): def __init__(self, genesis_state: LedgerState, config: Config):
self.config = config self.config = config
self.forks = [] self.forks = []
self.local_chain = Chain([], genesis=genesis_state.block) self.local_chain = genesis_state.block.id()
self.genesis_state = genesis_state self.genesis_state = genesis_state
self.ledger_state = {genesis_state.block: genesis_state.copy()} self.ledger_state = {genesis_state.block.id(): genesis_state.copy()}
self.epoch_state = {} self.epoch_state = {}
def validate_header(self, block: BlockHeader, chain: Chain) -> bool: def validate_header(self, block: BlockHeader) -> bool:
# TODO: verify blocks are not in the 'future' # TODO: verify blocks are not in the 'future'
if block.parent != chain.tip_id(): if block.parent not in self.ledger_state:
logger.warning("block parent is not chain tip") logger.warning("We have not seen block parent")
return False return False
current_state = self.ledger_state[block.parent].copy() current_state = self.ledger_state[block.parent].copy()
# We use the proposed block epoch state to validate orphans as well.
# For very old orphans, these states may be different.
epoch_state = self.compute_epoch_state(
block.slot.epoch(self.config), block.parent
)
# first, we verify adopted leadership transactions # first, we verify adopted leadership transactions
for orphan in block.orphaned_proofs: for orphan in block.orphaned_proofs:
# orphan proofs are checked in two ways # orphan proofs are checked in two ways
@ -410,10 +394,6 @@ class Follower:
logger.warning("missing orphan proof") logger.warning("missing orphan proof")
return False return False
# we use the proposed block epoch state here instead of the orphan's
# epoch state. For very old orphans, these states may be different.
epoch_state = self.compute_epoch_state(block.slot.epoch(self.config), chain)
# (2.) is satisfied by verifying the proof against current state ensuring: # (2.) is satisfied by verifying the proof against current state ensuring:
# - it is a valid proof # - it is a valid proof
# - and the nullifier has not already been spent # - and the nullifier has not already been spent
@ -431,7 +411,6 @@ class Follower:
# effects to the ledger state # effects to the ledger state
current_state.apply_leader_proof(orphan.leader_proof) current_state.apply_leader_proof(orphan.leader_proof)
epoch_state = self.compute_epoch_state(block.slot.epoch(self.config), chain)
# TODO: this is not the full block validation spec, only slot leader is verified # TODO: this is not the full block validation spec, only slot leader is verified
return self.verify_slot_leader( return self.verify_slot_leader(
block.slot, block.slot,
@ -462,129 +441,100 @@ class Follower:
and current_state.verify_unspent(proof.nullifier) and current_state.verify_unspent(proof.nullifier)
) )
# Try appending this block to an existing chain and return whether
# the operation was successful
def try_extend_chains(self, block: BlockHeader) -> Optional[Chain]:
if self.tip_id() == block.parent:
return self.local_chain
for chain in self.forks:
if chain.tip_id() == block.parent:
return chain
return None
def try_create_fork(self, block: BlockHeader) -> Optional[Chain]:
if self.genesis_state.block == block.parent:
# this block is forking off the genesis state
return Chain(blocks=[], genesis=self.genesis_state.block)
chains = self.forks + [self.local_chain]
for chain in chains:
block_position = chain.block_position(block.parent)
if block_position is not None:
return Chain(
blocks=chain.blocks[: block_position + 1],
genesis=self.genesis_state.block,
)
return None
def on_block(self, block: BlockHeader): def on_block(self, block: BlockHeader):
# check if the new block extends an existing chain if block.id() in self.ledger_state:
new_chain = self.try_extend_chains(block) logger.warning("dropping already processed block")
if new_chain is None:
# we failed to extend one of the existing chains,
# therefore we might need to create a new fork
new_chain = self.try_create_fork(block)
if new_chain is not None:
self.forks.append(new_chain)
else:
logger.warning("missing parent block")
# otherwise, we're missing the parent block
# in that case, just ignore the block
return
if not self.validate_header(block, new_chain):
logger.warning("invalid header")
return return
new_chain.blocks.append(block) if not self.validate_header(block):
logger.warning("invalid header")
# We may need to switch forks, lets run the fork choice rule to check. return
new_chain = self.fork_choice()
if new_chain != self.local_chain:
self.forks.remove(new_chain)
self.forks.append(self.local_chain)
self.local_chain = new_chain
new_state = self.ledger_state[block.parent].copy() new_state = self.ledger_state[block.parent].copy()
new_state.apply(block) new_state.apply(block)
self.ledger_state[block.id()] = new_state self.ledger_state[block.id()] = new_state
def unimported_orphans(self, tip: Id) -> list[BlockHeader]: if block.parent == self.local_chain:
# simply extending the local chain
self.local_chain = block.id()
else:
# otherwise, this block creates a fork
self.forks.append(block.id())
# remove any existing fork that is superceded by this block
if block.parent in self.forks:
self.forks.remove(block.parent)
# We may need to switch forks, lets run the fork choice rule to check.
new_tip = self.fork_choice()
self.forks.append(self.local_chain)
self.forks.remove(new_tip)
self.local_chain = new_tip
def unimported_orphans(self) -> list[BlockHeader]:
""" """
Returns all unimported orphans w.r.t. the given tip's state. Returns all unimported orphans w.r.t. the given tip's state.
Orphans are returned in the order that they should be imported. Orphans are returned in the order that they should be imported.
""" """
tip_state = self.ledger_state[tip].copy() tip_state = self.tip_state().copy()
tip = tip_state.block.id()
orphans = [] orphans = []
for fork in [self.local_chain, *self.forks]:
if fork.block_position(tip) is not None:
# the tip is a member of this fork, it doesn't make sense
# to take orphans from this fork as they are all already "imported"
continue
for block in fork.blocks: for fork in self.forks:
for b in [*block.orphaned_proofs, block]: _, fork_depth = common_prefix_depth(tip, fork, self.ledger_state)
if b.leader_proof.nullifier not in tip_state.nullifiers: for block_state in chain_suffix(fork, fork_depth, self.ledger_state):
tip_state.nullifiers.add(b.leader_proof.nullifier) b = block_state.block
orphans += [b] if b.leader_proof.nullifier not in tip_state.nullifiers:
tip_state.nullifiers.add(b.leader_proof.nullifier)
orphans += [b]
return orphans return orphans
# Evaluate the fork choice rule and return the chain we should be following # Evaluate the fork choice rule and return the chain we should be following
def fork_choice(self) -> Chain: def fork_choice(self) -> Id:
return maxvalid_bg( return maxvalid_bg(
self.local_chain, self.forks, k=self.config.k, s=self.config.s self.local_chain,
self.forks,
k=self.config.k,
s=self.config.s,
states=self.ledger_state,
) )
def tip(self) -> BlockHeader: def tip(self) -> BlockHeader:
return self.local_chain.tip() return self.tip_state().block
def tip_id(self) -> Id: def tip_id(self) -> Id:
return self.local_chain.tip_id() return self.local_chain
def tip_state(self) -> LedgerState: def tip_state(self) -> LedgerState:
return self.ledger_state[self.tip_id()] return self.ledger_state[self.tip_id()]
def state_at_slot_beginning(self, chain: Chain, slot: Slot) -> LedgerState: def state_at_slot_beginning(self, tip: Id, slot: Slot) -> LedgerState:
for block in reversed(chain.blocks): for state in iter_chain(tip, self.ledger_state):
if block.slot < slot: if state.block.slot < slot:
return self.ledger_state[block.id()] return state
return self.genesis_state return self.genesis_state
def epoch_start_slot(self, epoch) -> Slot: def epoch_start_slot(self, epoch) -> Slot:
return Slot(epoch.epoch * self.config.epoch_length) return Slot(epoch.epoch * self.config.epoch_length)
def stake_distribution_snapshot(self, epoch, chain): def stake_distribution_snapshot(self, epoch, tip: Id):
# stake distribution snapshot happens at the beginning of the previous epoch, # stake distribution snapshot happens at the beginning of the previous epoch,
# i.e. for epoch e, the snapshot is taken at the last block of epoch e-2 # i.e. for epoch e, the snapshot is taken at the last block of epoch e-2
slot = Slot(epoch.prev().epoch * self.config.epoch_length) slot = Slot(epoch.prev().epoch * self.config.epoch_length)
return self.state_at_slot_beginning(chain, slot) return self.state_at_slot_beginning(tip, slot)
def nonce_snapshot(self, epoch, chain): def nonce_snapshot(self, epoch, tip):
# nonce snapshot happens partway through the previous epoch after the # nonce snapshot happens partway through the previous epoch after the
# stake distribution has stabilized # stake distribution has stabilized
slot = Slot( slot = Slot(
self.config.epoch_relative_nonce_slot self.config.epoch_relative_nonce_slot
+ self.epoch_start_slot(epoch.prev()).absolute_slot + self.epoch_start_slot(epoch.prev()).absolute_slot
) )
return self.state_at_slot_beginning(chain, slot) return self.state_at_slot_beginning(tip, slot)
def compute_epoch_state(self, epoch: Epoch, chain: Chain) -> EpochState: def compute_epoch_state(self, epoch: Epoch, tip: Id) -> EpochState:
if epoch.epoch == 0: if epoch.epoch == 0:
return EpochState( return EpochState(
stake_distribution_snapshot=self.genesis_state, stake_distribution_snapshot=self.genesis_state,
@ -592,18 +542,18 @@ class Follower:
inferred_total_active_stake=self.config.initial_total_active_stake, inferred_total_active_stake=self.config.initial_total_active_stake,
) )
stake_distribution_snapshot = self.stake_distribution_snapshot(epoch, chain) stake_distribution_snapshot = self.stake_distribution_snapshot(epoch, tip)
nonce_snapshot = self.nonce_snapshot(epoch, chain) nonce_snapshot = self.nonce_snapshot(epoch, tip)
# we memoize epoch states to avoid recursion killing our performance # we memoize epoch states to avoid recursion killing our performance
memo_block_id = nonce_snapshot.block memo_block_id = nonce_snapshot.block.id()
if state := self.epoch_state.get((epoch, memo_block_id)): if state := self.epoch_state.get((epoch, memo_block_id)):
return state return state
# To update our inference of total stake, we need the prior estimate which # To update our inference of total stake, we need the prior estimate which
# was calculated last epoch. Thus we recurse here to retreive the previous # was calculated last epoch. Thus we recurse here to retreive the previous
# estimate of total stake. # estimate of total stake.
prev_epoch = self.compute_epoch_state(epoch.prev(), chain) prev_epoch = self.compute_epoch_state(epoch.prev(), tip)
inferred_total_active_stake = self._infer_total_active_stake( inferred_total_active_stake = self._infer_total_active_stake(
prev_epoch, nonce_snapshot, stake_distribution_snapshot prev_epoch, nonce_snapshot, stake_distribution_snapshot
) )
@ -696,45 +646,99 @@ class Leader:
) )
def common_prefix_len(a: Chain, b: Chain) -> int: def iter_chain(tip: Id, states: Dict[Id, LedgerState]):
for i, (x, y) in enumerate(zip(a.blocks, b.blocks)): while tip in states:
if x.id() != y.id(): yield states[tip]
return i tip = states[tip].block.parent
return min(len(a.blocks), len(b.blocks))
def chain_density(chain: Chain, slot: Slot) -> int: def chain_suffix(tip: Id, n: int, states: Dict[Id, LedgerState]) -> list[LedgerState]:
return len( return list(reversed(list(itertools.islice(iter_chain(tip, states), n))))
[
block
for block in chain.blocks
if block.slot.absolute_slot < slot.absolute_slot
]
)
# Implementation of the fork choice rule as defined in the Ouroboros Genesis paper def common_prefix_depth(a: Id, b: Id, states: Dict[Id, LedgerState]) -> (int, int):
# k defines the forking depth of chain we accept without more analysis a_blocks = iter_chain(a, states)
b_blocks = iter_chain(b, states)
seen = {}
depth = 0
while True:
try:
a_block = next(a_blocks).block.id()
if a_block in seen:
# we had seen this block from the fork chain
return depth, seen[a_block]
seen[a_block] = depth
except StopIteration:
pass
try:
b_block = next(b_blocks).block.id()
if b_block in seen:
# we had seen the fork in the local chain
return seen[b_block], depth
seen[b_block] = depth
except StopIteration:
pass
depth += 1
assert False
def chain_density(
head: Id, slot: Slot, reorg_depth: int, states: Dict[Id, LedgerState]
) -> int:
assert type(head) == Id
chain = iter_chain(head, states)
segment = itertools.islice(chain, reorg_depth)
return sum(1 for b in segment if b.block.slot < slot)
def block_children(states: Dict[Id, LedgerState]) -> Dict[Id, set[Id]]:
children = defaultdict(set)
for c, state in states.items():
children[state.block.parent].add(c)
return children
# Implementation of the Cryptarchia fork choice rule (following Ouroborous Genesis).
# The fork choice has two phases:
# 1. if the chain is not forking too deeply, we apply the longest chain fork choice rule
# 2. otherwise we look at the chain density immidiately following the fork
#
# k defines the forking depth of a chain at which point we switch phases.
# s defines the length of time (unit of slots) after the fork happened we will inspect for chain density # s defines the length of time (unit of slots) after the fork happened we will inspect for chain density
def maxvalid_bg(local_chain: Chain, forks: List[Chain], k: int, s: int) -> Chain: def maxvalid_bg(
local_chain: Id,
forks: List[Id],
k: int,
s: int,
states: Dict[Id, LedgerState],
) -> Id:
assert type(local_chain) == Id
assert all(type(f) == Id for f in forks)
cmax = local_chain cmax = local_chain
for chain in forks: for fork in forks:
lowest_common_ancestor = common_prefix_len(cmax, chain) cmax_depth, fork_depth = common_prefix_depth(cmax, fork, states)
m = cmax.length() - lowest_common_ancestor if cmax_depth <= k:
if m <= k: # Longest chain fork choice rule
# Classic longest chain rule with parameter k if cmax_depth < fork_depth:
if cmax.length() < chain.length(): cmax = fork
cmax = chain
else: else:
# The chain is forking too much, we need to pay a bit more attention # The chain is forking too much, we need to pay a bit more attention
# In particular, select the chain that is the densest after the fork # In particular, select the chain that is the densest after the fork
forking_slot = Slot( cmax_divergent_block = chain_suffix(cmax, cmax_depth, states)[0].block
cmax.blocks[lowest_common_ancestor].slot.absolute_slot + s
) forking_slot = Slot(cmax_divergent_block.slot.absolute_slot + s)
cmax_density = chain_density(cmax, forking_slot) cmax_density = chain_density(cmax, forking_slot, cmax_depth, states)
candidate_density = chain_density(chain, forking_slot) fork_density = chain_density(fork, forking_slot, fork_depth, states)
if cmax_density < candidate_density:
cmax = chain if cmax_density < fork_density:
cmax = fork
return cmax return cmax

33
cryptarchia/test_common.py
View File

@ -1,7 +1,5 @@
from .cryptarchia import ( from .cryptarchia import (
Config, Config,
TimeConfig,
Id,
Slot, Slot,
Coin, Coin,
BlockHeader, BlockHeader,
@ -20,19 +18,18 @@ class TestNode:
def epoch_state(self, slot: Slot): def epoch_state(self, slot: Slot):
return self.follower.compute_epoch_state( return self.follower.compute_epoch_state(
slot.epoch(self.config), self.follower.local_chain slot.epoch(self.config), self.follower.tip_id()
) )
def on_slot(self, slot: Slot) -> BlockHeader | None: def on_slot(self, slot: Slot) -> BlockHeader | None:
parent = self.follower.tip_id() parent = self.follower.tip_id()
epoch_state = self.epoch_state(slot) epoch_state = self.epoch_state(slot)
if leader_proof := self.leader.try_prove_slot_leader(epoch_state, slot, parent): if leader_proof := self.leader.try_prove_slot_leader(epoch_state, slot, parent):
orphans = self.follower.unimported_orphans(parent)
self.leader.coin = self.leader.coin.evolve() self.leader.coin = self.leader.coin.evolve()
return BlockHeader( return BlockHeader(
parent=parent, parent=parent,
slot=slot, slot=slot,
orphaned_proofs=orphans, orphaned_proofs=self.follower.unimported_orphans(),
leader_proof=leader_proof, leader_proof=leader_proof,
content_size=0, content_size=0,
content_id=bytes(32), content_id=bytes(32),
@ -53,7 +50,15 @@ def mk_config(initial_stake_distribution: list[Coin]) -> Config:
def mk_genesis_state(initial_stake_distribution: list[Coin]) -> LedgerState: def mk_genesis_state(initial_stake_distribution: list[Coin]) -> LedgerState:
return LedgerState( return LedgerState(
block=bytes(32), block=BlockHeader(
slot=Slot(0),
parent=bytes(32),
content_size=0,
content_id=bytes(32),
leader_proof=MockLeaderProof.new(
Coin(sk=0, value=0), Slot(0), parent=bytes(32)
),
),
nonce=bytes(32), nonce=bytes(32),
commitments_spend={c.commitment() 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}, commitments_lead={c.commitment() for c in initial_stake_distribution},
@ -62,26 +67,30 @@ def mk_genesis_state(initial_stake_distribution: list[Coin]) -> LedgerState:
def mk_block( def mk_block(
parent: Id, slot: int, coin: Coin, content=bytes(32), orphaned_proofs=[] parent: BlockHeader, slot: int, coin: Coin, content=bytes(32), orphaned_proofs=[]
) -> BlockHeader: ) -> BlockHeader:
assert len(parent) == 32 assert type(parent) == BlockHeader, type(parent)
assert type(slot) == int, type(slot)
from hashlib import sha256 from hashlib import sha256
return BlockHeader( return BlockHeader(
slot=Slot(slot), slot=Slot(slot),
parent=parent, parent=parent.id(),
content_size=len(content), content_size=len(content),
content_id=sha256(content).digest(), content_id=sha256(content).digest(),
leader_proof=MockLeaderProof.new(coin, Slot(slot), parent=parent), leader_proof=MockLeaderProof.new(coin, Slot(slot), parent=parent.id()),
orphaned_proofs=orphaned_proofs, orphaned_proofs=orphaned_proofs,
) )
def mk_chain(parent, coin: Coin, slots: list[int]) -> tuple[list[BlockHeader], Coin]: def mk_chain(
parent: BlockHeader, coin: Coin, slots: list[int]
) -> tuple[list[BlockHeader], Coin]:
assert type(parent) == BlockHeader
chain = [] chain = []
for s in slots: for s in slots:
block = mk_block(parent=parent, slot=s, coin=coin) block = mk_block(parent=parent, slot=s, coin=coin)
chain.append(block) chain.append(block)
parent = block.id() parent = block
coin = coin.evolve() coin = coin.evolve()
return chain, coin return chain, coin

109
cryptarchia/test_fork_choice.py
View File

@ -1,40 +1,82 @@
from unittest import TestCase from unittest import TestCase
from itertools import repeat
import numpy as np
import hashlib
from copy import deepcopy from copy import deepcopy
from cryptarchia.cryptarchia import ( from cryptarchia.cryptarchia import (
maxvalid_bg, maxvalid_bg,
Chain,
BlockHeader,
Slot, Slot,
Id,
MockLeaderProof,
Coin, Coin,
Follower, Follower,
common_prefix_depth,
LedgerState,
) )
from .test_common import mk_chain, mk_config, mk_genesis_state, mk_block from .test_common import mk_chain, mk_config, mk_genesis_state, mk_block
class TestForkChoice(TestCase): class TestForkChoice(TestCase):
def test_common_prefix_depth(self):
# 6 - 7
# /
# 0 - 1 - 2 - 3
# \
# 4 - 5
coin = Coin(sk=1, value=100)
b0 = mk_genesis_state([]).block
b1 = mk_block(b0, 1, coin)
b2 = mk_block(b1, 2, coin)
b3 = mk_block(b2, 3, coin)
b4 = mk_block(b0, 1, coin, content=b"b4")
b5 = mk_block(b4, 2, coin)
b6 = mk_block(b2, 3, coin, content=b"b6")
b7 = mk_block(b6, 4, coin)
states = {
b.id(): LedgerState(block=b) for b in [b0, b1, b2, b3, b4, b5, b6, b7]
}
assert (d := common_prefix_depth(b0.id(), b0.id(), states)) == (0, 0), d
assert (d := common_prefix_depth(b1.id(), b0.id(), states)) == (1, 0), d
assert (d := common_prefix_depth(b0.id(), b1.id(), states)) == (0, 1), d
assert (d := common_prefix_depth(b1.id(), b1.id(), states)) == (0, 0), d
assert (d := common_prefix_depth(b2.id(), b0.id(), states)) == (2, 0), d
assert (d := common_prefix_depth(b0.id(), b2.id(), states)) == (0, 2), d
assert (d := common_prefix_depth(b3.id(), b0.id(), states)) == (3, 0), d
assert (d := common_prefix_depth(b0.id(), b3.id(), states)) == (0, 3), d
assert (d := common_prefix_depth(b1.id(), b4.id(), states)) == (1, 1), d
assert (d := common_prefix_depth(b4.id(), b1.id(), states)) == (1, 1), d
assert (d := common_prefix_depth(b1.id(), b5.id(), states)) == (1, 2), d
assert (d := common_prefix_depth(b5.id(), b1.id(), states)) == (2, 1), d
assert (d := common_prefix_depth(b2.id(), b5.id(), states)) == (2, 2), d
assert (d := common_prefix_depth(b5.id(), b2.id(), states)) == (2, 2), d
assert (d := common_prefix_depth(b3.id(), b5.id(), states)) == (3, 2), d
assert (d := common_prefix_depth(b5.id(), b3.id(), states)) == (2, 3), d
assert (d := common_prefix_depth(b3.id(), b6.id(), states)) == (1, 1), d
assert (d := common_prefix_depth(b6.id(), b3.id(), states)) == (1, 1), d
assert (d := common_prefix_depth(b3.id(), b7.id(), states)) == (1, 2), d
assert (d := common_prefix_depth(b7.id(), b3.id(), states)) == (2, 1), d
assert (d := common_prefix_depth(b5.id(), b7.id(), states)) == (2, 4), d
assert (d := common_prefix_depth(b7.id(), b5.id(), states)) == (4, 2), d
def test_fork_choice_long_sparse_chain(self): def test_fork_choice_long_sparse_chain(self):
# The longest chain is not dense after the fork # The longest chain is not dense after the fork
genesis = mk_genesis_state([]).block
short_coin, long_coin = Coin(sk=0, value=100), Coin(sk=1, value=100) short_coin, long_coin = Coin(sk=0, value=100), Coin(sk=1, value=100)
common, long_coin = mk_chain(parent=bytes(32), coin=long_coin, slots=range(50)) common, long_coin = mk_chain(parent=genesis, coin=long_coin, slots=range(50))
long_chain_sparse_ext, long_coin = mk_chain( long_chain_sparse_ext, long_coin = mk_chain(
parent=common[-1].id(), coin=long_coin, slots=range(50, 100, 2) parent=common[-1], coin=long_coin, slots=range(50, 100, 2)
) )
short_chain_dense_ext, _ = mk_chain( short_chain_dense_ext, _ = mk_chain(
parent=common[-1].id(), coin=short_coin, slots=range(50, 100) parent=common[-1], coin=short_coin, slots=range(50, 100)
) )
# add more blocks to the long chain to ensure the long chain is indeed longer # add more blocks to the long chain to ensure the long chain is indeed longer
long_chain_further_ext, _ = mk_chain( long_chain_further_ext, _ = mk_chain(
parent=long_chain_sparse_ext[-1].id(), coin=long_coin, slots=range(100, 126) parent=long_chain_sparse_ext[-1], coin=long_coin, slots=range(100, 126)
) )
long_chain = deepcopy(common) + long_chain_sparse_ext + long_chain_further_ext long_chain = deepcopy(common) + long_chain_sparse_ext + long_chain_further_ext
@ -45,26 +87,34 @@ class TestForkChoice(TestCase):
k = 1 k = 1
s = 50 s = 50
short_chain = Chain(short_chain, genesis=bytes(32)) states = {b.id(): LedgerState(block=b) for b in short_chain + long_chain}
long_chain = Chain(long_chain, genesis=bytes(32))
assert maxvalid_bg(short_chain, [long_chain], k, s) == short_chain assert (
maxvalid_bg(short_chain[-1].id(), [long_chain[-1].id()], k, s, states)
== short_chain[-1].id()
)
# However, if we set k to the fork length, it will be accepted # However, if we set k to the fork length, it will be accepted
k = long_chain.length() k = len(long_chain)
assert maxvalid_bg(short_chain, [long_chain], k, s) == long_chain assert (
maxvalid_bg(short_chain[-1].id(), [long_chain[-1].id()], k, s, states)
== long_chain[-1].id()
)
def test_fork_choice_long_dense_chain(self): def test_fork_choice_long_dense_chain(self):
# The longest chain is also the densest after the fork # The longest chain is also the densest after the fork
short_coin, long_coin = Coin(sk=0, value=100), Coin(sk=1, value=100) short_coin, long_coin = Coin(sk=0, value=100), Coin(sk=1, value=100)
common, long_coin = mk_chain( common, long_coin = mk_chain(
parent=bytes(32), coin=long_coin, slots=range(1, 50) parent=mk_genesis_state([]).block,
coin=long_coin,
slots=range(1, 50),
) )
long_chain_dense_ext, _ = mk_chain( long_chain_dense_ext, _ = mk_chain(
parent=common[-1].id(), coin=long_coin, slots=range(50, 100) parent=common[-1], coin=long_coin, slots=range(50, 100)
) )
short_chain_sparse_ext, _ = mk_chain( short_chain_sparse_ext, _ = mk_chain(
parent=common[-1].id(), coin=short_coin, slots=range(50, 100, 2) parent=common[-1], coin=short_coin, slots=range(50, 100, 2)
) )
long_chain = deepcopy(common) + long_chain_dense_ext long_chain = deepcopy(common) + long_chain_dense_ext
@ -72,9 +122,12 @@ class TestForkChoice(TestCase):
k = 1 k = 1
s = 50 s = 50
short_chain = Chain(short_chain, genesis=bytes(32)) states = {b.id(): LedgerState(block=b) for b in short_chain + long_chain}
long_chain = Chain(long_chain, genesis=bytes(32))
assert maxvalid_bg(short_chain, [long_chain], k, s) == long_chain assert (
maxvalid_bg(short_chain[-1].id(), [long_chain[-1].id()], k, s, states)
== long_chain[-1].id()
)
def test_fork_choice_integration(self): def test_fork_choice_integration(self):
c_a, c_b = Coin(sk=0, value=10), Coin(sk=1, value=10) c_a, c_b = Coin(sk=0, value=10), Coin(sk=1, value=10)
@ -88,7 +141,7 @@ class TestForkChoice(TestCase):
follower.on_block(b1) follower.on_block(b1)
assert follower.tip_id() == b1.id() assert follower.tip_id() == b1.id()
assert follower.forks == [] assert follower.forks == [], follower.forks
# -- then we fork -- # -- then we fork --
# #
@ -99,14 +152,14 @@ class TestForkChoice(TestCase):
# b3 # b3
# #
b2, c_a = mk_block(b1.id(), 2, c_a), c_a.evolve() b2, c_a = mk_block(b1, 2, c_a), c_a.evolve()
b3, c_b = mk_block(b1.id(), 2, c_b), c_b.evolve() b3, c_b = mk_block(b1, 2, c_b), c_b.evolve()
follower.on_block(b2) follower.on_block(b2)
follower.on_block(b3) follower.on_block(b3)
assert follower.tip_id() == b2.id() assert follower.tip_id() == b2.id()
assert len(follower.forks) == 1 and follower.forks[0].tip_id() == b3.id() assert len(follower.forks) == 1 and follower.forks[0] == b3.id()
# -- extend the fork causing a re-org -- # -- extend the fork causing a re-org --
# #
@ -117,8 +170,8 @@ class TestForkChoice(TestCase):
# b3 - b4 == tip # b3 - b4 == tip
# #
b4, c_b = mk_block(b3.id(), 3, c_b), c_a.evolve() b4, c_b = mk_block(b3, 3, c_b), c_a.evolve()
follower.on_block(b4) follower.on_block(b4)
assert follower.tip_id() == b4.id() assert follower.tip_id() == b4.id()
assert len(follower.forks) == 1 and follower.forks[0].tip_id() == b2.id() assert len(follower.forks) == 1 and follower.forks[0] == b2.id(), follower.forks

15
cryptarchia/test_leader.py
View File

@ -2,24 +2,15 @@ from unittest import TestCase
import numpy as np import numpy as np
from .cryptarchia import ( from .cryptarchia import Leader, EpochState, LedgerState, Coin, phi, Slot
Leader,
Config,
EpochState,
LedgerState,
Coin,
phi,
TimeConfig,
Slot,
)
from .test_common import mk_config from .test_common import mk_config
class TestLeader(TestCase): class TestLeader(TestCase):
def test_slot_leader_statistics(self): def test_slot_leader_statistics(self):
epoch = EpochState( epoch = EpochState(
stake_distribution_snapshot=LedgerState(), stake_distribution_snapshot=LedgerState(block=None),
nonce_snapshot=LedgerState(nonce=b"1010101010"), nonce_snapshot=LedgerState(block=None, nonce=b"1010101010"),
inferred_total_active_stake=1000, inferred_total_active_stake=1000,
) )

93
cryptarchia/test_ledger_state_update.py
View File

@ -2,22 +2,33 @@ from unittest import TestCase
import numpy as np import numpy as np
from .cryptarchia import ( from .cryptarchia import Follower, Coin, iter_chain
Follower,
TimeConfig,
BlockHeader,
Config,
Coin,
LedgerState,
MockLeaderProof,
Slot,
Id,
)
from .test_common import mk_config, mk_block, mk_genesis_state from .test_common import mk_config, mk_block, mk_genesis_state
class TestLedgerStateUpdate(TestCase): class TestLedgerStateUpdate(TestCase):
def test_on_block_idempotent(self):
leader_coin = Coin(sk=0, value=100)
genesis = mk_genesis_state([leader_coin])
follower = Follower(genesis, mk_config([leader_coin]))
block = mk_block(slot=0, parent=genesis.block, coin=leader_coin)
follower.on_block(block)
# Follower should have accepted the block
assert len(list(iter_chain(follower.tip_id(), follower.ledger_state))) == 2
assert follower.tip() == block
follower.on_block(block)
# Should have been a No-op
assert len(list(iter_chain(follower.tip_id(), follower.ledger_state))) == 2
assert follower.tip() == block
assert len(follower.ledger_state) == 2
assert len(follower.forks) == 0
def test_ledger_state_prevents_coin_reuse(self): def test_ledger_state_prevents_coin_reuse(self):
leader_coin = Coin(sk=0, value=100) leader_coin = Coin(sk=0, value=100)
genesis = mk_genesis_state([leader_coin]) genesis = mk_genesis_state([leader_coin])
@ -28,17 +39,17 @@ class TestLedgerStateUpdate(TestCase):
follower.on_block(block) follower.on_block(block)
# Follower should have accepted the block # Follower should have accepted the block
assert follower.local_chain.length() == 1 assert len(list(iter_chain(follower.tip_id(), follower.ledger_state))) == 2
assert follower.tip() == block assert follower.tip() == block
# Follower should have updated their ledger state to mark the leader coin as spent # Follower should have updated their ledger state to mark the leader coin as spent
assert follower.tip_state().verify_unspent(leader_coin.nullifier()) == False assert follower.tip_state().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, coin=leader_coin)
follower.on_block(block) follower.on_block(reuse_coin_block)
# Follower should *not* have accepted the block # Follower should *not* have accepted the block
assert follower.local_chain.length() == 1 assert len(list(iter_chain(follower.tip_id(), follower.ledger_state))) == 2
assert follower.tip() == block assert follower.tip() == block
def test_ledger_state_is_properly_updated_on_reorg(self): def test_ledger_state_is_properly_updated_on_reorg(self):
@ -67,7 +78,7 @@ class TestLedgerStateUpdate(TestCase):
# 4) then coin[2] wins slot 1 and chooses to extend from block_2 # 4) then coin[2] wins slot 1 and chooses to extend from block_2
block_3 = mk_block(parent=block_2.id(), slot=1, coin=coin[2]) block_3 = mk_block(parent=block_2, slot=1, coin=coin[2])
follower.on_block(block_3) follower.on_block(block_3)
# the follower should have switched over to the block_2 fork # the follower should have switched over to the block_2 fork
assert follower.tip() == block_3 assert follower.tip() == block_3
@ -89,37 +100,37 @@ class TestLedgerStateUpdate(TestCase):
follower.on_block(block_2) follower.on_block(block_2)
assert follower.tip() == block_1 assert follower.tip() == block_1
assert len(follower.forks) == 1, f"{len(follower.forks)}" assert len(follower.forks) == 1, f"{len(follower.forks)}"
assert follower.forks[0].tip() == block_2 assert follower.forks[0] == block_2.id()
# coin_2 wins slot 1 and chooses to extend from block_1 # coin_2 wins slot 1 and chooses to extend from block_1
# coin_3 also wins slot 1 and but chooses to extend from block_2 # coin_3 also wins slot 1 and but chooses to extend from block_2
# Both blocks are accepted. Both the local chain and the fork grow. No fork is newly created. # Both blocks are accepted. Both the local chain and the fork grow. No fork is newly created.
block_3 = mk_block(parent=block_1.id(), slot=1, coin=coins[2]) block_3 = mk_block(parent=block_1, slot=1, coin=coins[2])
block_4 = mk_block(parent=block_2.id(), slot=1, coin=coins[3]) block_4 = mk_block(parent=block_2, slot=1, coin=coins[3])
follower.on_block(block_3) follower.on_block(block_3)
follower.on_block(block_4) follower.on_block(block_4)
assert follower.tip() == block_3 assert follower.tip() == block_3
assert len(follower.forks) == 1, f"{len(follower.forks)}" assert len(follower.forks) == 1, f"{len(follower.forks)}"
assert follower.forks[0].tip() == block_4 assert follower.forks[0] == block_4.id()
# coin_4 wins slot 1 and but chooses to extend from block_2 as well # coin_4 wins slot 1 and but chooses to extend from block_2 as well
# The block is accepted. A new fork is created "from the block_2". # The block is accepted. A new fork is created "from the block_2".
block_5 = mk_block(parent=block_2.id(), slot=1, coin=coins[4]) block_5 = mk_block(parent=block_2, slot=1, coin=coins[4])
follower.on_block(block_5) follower.on_block(block_5)
assert follower.tip() == block_3 assert follower.tip() == block_3
assert len(follower.forks) == 2, f"{len(follower.forks)}" assert len(follower.forks) == 2, f"{len(follower.forks)}"
assert follower.forks[0].tip() == block_4 assert follower.forks[0] == block_4.id()
assert follower.forks[1].tip() == block_5 assert follower.forks[1] == block_5.id()
# A block based on an unknown parent is not accepted. # A block based on an unknown parent is not accepted.
# Nothing changes from the local chain and forks. # Nothing changes from the local chain and forks.
unknown_block = mk_block(parent=block_5.id(), slot=2, coin=coins[5]) unknown_block = mk_block(parent=block_5, slot=2, coin=coins[5])
block_6 = mk_block(parent=unknown_block.id(), slot=2, coin=coins[6]) block_6 = mk_block(parent=unknown_block, slot=2, coin=coins[6])
follower.on_block(block_6) follower.on_block(block_6)
assert follower.tip() == block_3 assert follower.tip() == block_3
assert len(follower.forks) == 2, f"{len(follower.forks)}" assert len(follower.forks) == 2, f"{len(follower.forks)}"
assert follower.forks[0].tip() == block_4 assert follower.forks[0] == block_4.id()
assert follower.forks[1].tip() == block_5 assert follower.forks[1] == block_5.id()
def test_epoch_transition(self): def test_epoch_transition(self):
leader_coins = [Coin(sk=i, value=100) for i in range(4)] leader_coins = [Coin(sk=i, value=100) for i in range(4)]
@ -138,14 +149,14 @@ class TestLedgerStateUpdate(TestCase):
assert follower.tip() == block_1 assert follower.tip() == block_1
assert follower.tip().slot.epoch(config).epoch == 0 assert follower.tip().slot.epoch(config).epoch == 0
block_2 = mk_block(slot=19, parent=block_1.id(), coin=leader_coins[1]) block_2 = mk_block(slot=19, parent=block_1, coin=leader_coins[1])
follower.on_block(block_2) follower.on_block(block_2)
assert follower.tip() == block_2 assert follower.tip() == block_2
assert follower.tip().slot.epoch(config).epoch == 0 assert follower.tip().slot.epoch(config).epoch == 0
# ---- EPOCH 1 ---- # ---- EPOCH 1 ----
block_3 = mk_block(slot=20, parent=block_2.id(), coin=leader_coins[2]) block_3 = mk_block(slot=20, parent=block_2, coin=leader_coins[2])
follower.on_block(block_3) follower.on_block(block_3)
assert follower.tip() == block_3 assert follower.tip() == block_3
assert follower.tip().slot.epoch(config).epoch == 1 assert follower.tip().slot.epoch(config).epoch == 1
@ -157,7 +168,7 @@ class TestLedgerStateUpdate(TestCase):
# To ensure this is the case, we add a new coin just to the state associated with that slot, # 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 # 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 # first, verify that if we don't change the state, the block is not accepted
block_4 = mk_block(slot=40, parent=block_3.id(), coin=Coin(sk=4, value=100)) block_4 = mk_block(slot=40, parent=block_3, coin=Coin(sk=4, value=100))
follower.on_block(block_4) follower.on_block(block_4)
assert follower.tip() == block_3 assert follower.tip() == block_3
# then we add the coin to "spendable commitments" associated with slot 9 # then we add the coin to "spendable commitments" associated with slot 9
@ -181,12 +192,12 @@ class TestLedgerStateUpdate(TestCase):
assert follower.tip() == block_1 assert follower.tip() == block_1
# coin can't be reused to win following slots: # coin can't be reused to win following slots:
block_2_reuse = mk_block(slot=1, parent=block_1.id(), coin=coin) block_2_reuse = mk_block(slot=1, parent=block_1, coin=coin)
follower.on_block(block_2_reuse) follower.on_block(block_2_reuse)
assert follower.tip() == block_1 assert follower.tip() == block_1
# but the evolved coin is eligible # but the evolved coin is eligible
block_2_evolve = mk_block(slot=1, parent=block_1.id(), coin=coin.evolve()) block_2_evolve = mk_block(slot=1, parent=block_1, coin=coin.evolve())
follower.on_block(block_2_evolve) follower.on_block(block_2_evolve)
assert follower.tip() == block_2_evolve assert follower.tip() == block_2_evolve
@ -212,12 +223,12 @@ class TestLedgerStateUpdate(TestCase):
) )
# the new coin is not yet eligible for elections # 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) block_0_1_attempt = mk_block(slot=1, parent=block_0_0, coin=coin_new)
follower.on_block(block_0_1_attempt) follower.on_block(block_0_1_attempt)
assert follower.tip() == block_0_0 assert follower.tip() == block_0_0
# whereas the evolved coin from genesis can be spent immediately # 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()) block_0_1 = mk_block(slot=1, parent=block_0_0, coin=coin.evolve())
follower.on_block(block_0_1) follower.on_block(block_0_1)
assert follower.tip() == block_0_1 assert follower.tip() == block_0_1
@ -226,7 +237,7 @@ class TestLedgerStateUpdate(TestCase):
# The newly minted coin is still not eligible in the following epoch since the # The newly minted coin is still not eligible in the following epoch since the
# stake distribution snapshot is taken at the beginning of the previous epoch # stake distribution snapshot is taken at the beginning of the previous epoch
block_1_0 = mk_block(slot=20, parent=block_0_1.id(), coin=coin_new) block_1_0 = mk_block(slot=20, parent=block_0_1, coin=coin_new)
follower.on_block(block_1_0) follower.on_block(block_1_0)
assert follower.tip() == block_0_1 assert follower.tip() == block_0_1
@ -234,16 +245,12 @@ class TestLedgerStateUpdate(TestCase):
# The coin is finally eligible 2 epochs after it was first minted # The coin is finally eligible 2 epochs after it was first minted
block_2_0 = mk_block( block_2_0 = mk_block(slot=40, parent=block_0_1, coin=coin_new)
slot=40,
parent=block_0_1.id(),
coin=coin_new,
)
follower.on_block(block_2_0) follower.on_block(block_2_0)
assert follower.tip() == block_2_0 assert follower.tip() == block_2_0
# And now the minted coin can freely use the evolved coin for subsequent blocks # And now the minted coin can freely use the evolved coin for subsequent blocks
block_2_1 = mk_block(slot=40, parent=block_2_0.id(), coin=coin_new.evolve()) block_2_1 = mk_block(slot=40, parent=block_2_0, coin=coin_new.evolve())
follower.on_block(block_2_1) follower.on_block(block_2_1)
assert follower.tip() == block_2_1 assert follower.tip() == block_2_1
@ -259,7 +266,7 @@ class TestLedgerStateUpdate(TestCase):
coin_new = coin.evolve() coin_new = coin.evolve()
coin_new_new = coin_new.evolve() coin_new_new = coin_new.evolve()
block_0_1 = mk_block(slot=1, parent=block_0_0.id(), coin=coin_new_new) block_0_1 = mk_block(slot=1, parent=block_0_0, coin=coin_new_new)
follower.on_block(block_0_1) follower.on_block(block_0_1)
# the coin evolved twice should not be accepted as it is not in the lead commitments # the coin evolved twice should not be accepted as it is not in the lead commitments
assert follower.tip() == block_0_0 assert follower.tip() == block_0_0
@ -272,7 +279,7 @@ class TestLedgerStateUpdate(TestCase):
orphan = mk_block(parent=genesis.block, slot=0, coin=coin_orphan) orphan = mk_block(parent=genesis.block, slot=0, coin=coin_orphan)
block_0_1 = mk_block( block_0_1 = mk_block(
slot=1, slot=1,
parent=block_0_0.id(), parent=block_0_0,
coin=coin_orphan.evolve(), coin=coin_orphan.evolve(),
orphaned_proofs=[orphan], orphaned_proofs=[orphan],
) )

105
cryptarchia/test_orphaned_proofs.py
View File

@ -1,21 +1,8 @@
from unittest import TestCase from unittest import TestCase
from itertools import repeat
import numpy as np
import hashlib
from copy import deepcopy from cryptarchia.cryptarchia import Coin, Follower
from cryptarchia.cryptarchia import (
maxvalid_bg,
Chain,
BlockHeader,
Slot,
Id,
MockLeaderProof,
Coin,
Follower,
)
from .test_common import mk_chain, mk_config, mk_genesis_state, mk_block from .test_common import mk_config, mk_genesis_state, mk_block
class TestOrphanedProofs(TestCase): class TestOrphanedProofs(TestCase):
@ -36,15 +23,15 @@ class TestOrphanedProofs(TestCase):
# #
b1, c_a = mk_block(genesis.block, 1, c_a), c_a.evolve() b1, c_a = mk_block(genesis.block, 1, c_a), c_a.evolve()
b2, c_a = mk_block(b1.id(), 2, c_a), c_a.evolve() b2, c_a = mk_block(b1, 2, c_a), c_a.evolve()
b3, c_b = mk_block(b1.id(), 2, c_b), c_b.evolve() b3, c_b = mk_block(b1, 2, c_b), c_b.evolve()
for b in [b1, b2, b3]: for b in [b1, b2, b3]:
follower.on_block(b) follower.on_block(b)
assert follower.tip() == b2 assert follower.tip() == b2
assert [f.tip() for f in follower.forks] == [b3] assert [f for f in follower.forks] == [b3.id()]
assert follower.unimported_orphans(follower.tip_id()) == [b3] assert follower.unimported_orphans() == [b3]
# -- extend with import -- # -- extend with import --
# #
@ -54,12 +41,12 @@ class TestOrphanedProofs(TestCase):
# \ / # \ /
# b3 # b3
# #
b4, c_a = mk_block(b2.id(), 3, c_a, orphaned_proofs=[b3]), c_a.evolve() b4, c_a = mk_block(b2, 3, c_a, orphaned_proofs=[b3]), c_a.evolve()
follower.on_block(b4) follower.on_block(b4)
assert follower.tip() == b4 assert follower.tip() == b4
assert [f.tip() for f in follower.forks] == [b3] assert [f for f in follower.forks] == [b3.id()]
assert follower.unimported_orphans(follower.tip_id()) == [] assert follower.unimported_orphans() == []
def test_orphan_proof_import_from_long_running_fork(self): def test_orphan_proof_import_from_long_running_fork(self):
c_a, c_b = Coin(sk=0, value=10), Coin(sk=1, value=10) c_a, c_b = Coin(sk=0, value=10), Coin(sk=1, value=10)
@ -79,18 +66,18 @@ class TestOrphanedProofs(TestCase):
b1, c_a = mk_block(genesis.block, 1, c_a), c_a.evolve() b1, c_a = mk_block(genesis.block, 1, c_a), c_a.evolve()
b2, c_a = mk_block(b1.id(), 2, c_a), c_a.evolve() b2, c_a = mk_block(b1, 2, c_a), c_a.evolve()
b3, c_a = mk_block(b2.id(), 3, c_a), c_a.evolve() b3, c_a = mk_block(b2, 3, c_a), c_a.evolve()
b4, c_b = mk_block(b1.id(), 2, c_b), c_b.evolve() b4, c_b = mk_block(b1, 2, c_b), c_b.evolve()
b5, c_b = mk_block(b4.id(), 3, c_b), c_b.evolve() b5, c_b = mk_block(b4, 3, c_b), c_b.evolve()
for b in [b1, b2, b3, b4, b5]: for b in [b1, b2, b3, b4, b5]:
follower.on_block(b) follower.on_block(b)
assert follower.tip() == b3 assert follower.tip() == b3
assert [f.tip() for f in follower.forks] == [b5] assert [f for f in follower.forks] == [b5.id()]
assert follower.unimported_orphans(follower.tip_id()) == [b4, b5] assert follower.unimported_orphans() == [b4, b5]
# -- extend b3, importing the fork -- # -- extend b3, importing the fork --
# #
@ -100,11 +87,11 @@ class TestOrphanedProofs(TestCase):
# \ / / # \ / /
# b4 - b5 # b4 - b5
b6, c_a = mk_block(b3.id(), 4, c_a, orphaned_proofs=[b4, b5]), c_a.evolve() b6, c_a = mk_block(b3, 4, c_a, orphaned_proofs=[b4, b5]), c_a.evolve()
follower.on_block(b6) follower.on_block(b6)
assert follower.tip() == b6 assert follower.tip() == b6
assert [f.tip() for f in follower.forks] == [b5] assert [f for f in follower.forks] == [b5.id()]
def test_orphan_proof_import_from_fork_without_direct_shared_parent(self): def test_orphan_proof_import_from_fork_without_direct_shared_parent(self):
coins = [Coin(sk=i, value=10) for i in range(2)] coins = [Coin(sk=i, value=10) for i in range(2)]
@ -123,20 +110,20 @@ class TestOrphanedProofs(TestCase):
b1, c_a = mk_block(genesis.block, 1, c_a), c_a.evolve() b1, c_a = mk_block(genesis.block, 1, c_a), c_a.evolve()
b2, c_a = mk_block(b1.id(), 2, c_a), c_a.evolve() b2, c_a = mk_block(b1, 2, c_a), c_a.evolve()
b3, c_a = mk_block(b2.id(), 3, c_a), c_a.evolve() b3, c_a = mk_block(b2, 3, c_a), c_a.evolve()
b4, c_a = mk_block(b3.id(), 4, c_a), c_a.evolve() b4, c_a = mk_block(b3, 4, c_a), c_a.evolve()
b5, c_b = mk_block(b1.id(), 2, c_b), c_b.evolve() b5, c_b = mk_block(b1, 2, c_b), c_b.evolve()
b6, c_b = mk_block(b5.id(), 3, c_b), c_b.evolve() b6, c_b = mk_block(b5, 3, c_b), c_b.evolve()
b7, c_b = mk_block(b6.id(), 4, c_b), c_b.evolve() b7, c_b = mk_block(b6, 4, c_b), c_b.evolve()
for b in [b1, b2, b3, b4, b5, b6, b7]: for b in [b1, b2, b3, b4, b5, b6, b7]:
follower.on_block(b) follower.on_block(b)
assert follower.tip() == b4 assert follower.tip() == b4
assert [f.tip() for f in follower.forks] == [b7] assert [f for f in follower.forks] == [b7.id()]
assert follower.unimported_orphans(follower.tip_id()) == [b5, b6, b7] assert follower.unimported_orphans() == [b5, b6, b7]
# -- extend b4, importing the forks -- # -- extend b4, importing the forks --
# #
@ -149,12 +136,12 @@ class TestOrphanedProofs(TestCase):
# Earlier implementations of orphan proof validation failed to # Earlier implementations of orphan proof validation failed to
# validate b7 as an orphan here. # validate b7 as an orphan here.
b8, c_a = mk_block(b4.id(), 5, c_a, orphaned_proofs=[b5, b6, b7]), c_a.evolve() b8, c_a = mk_block(b4, 5, c_a, orphaned_proofs=[b5, b6, b7]), c_a.evolve()
follower.on_block(b8) follower.on_block(b8)
assert follower.tip() == b8 assert follower.tip() == b8
assert [f.tip() for f in follower.forks] == [b7] assert [f for f in follower.forks] == [b7.id()]
assert follower.unimported_orphans(follower.tip_id()) == [] assert follower.unimported_orphans() == []
def test_unimported_orphans(self): def test_unimported_orphans(self):
# Given the following fork graph: # Given the following fork graph:
@ -187,22 +174,22 @@ class TestOrphanedProofs(TestCase):
b1, c_a = mk_block(genesis.block, 1, c_a), c_a.evolve() b1, c_a = mk_block(genesis.block, 1, c_a), c_a.evolve()
b2, c_a = mk_block(b1.id(), 2, c_a), c_a.evolve() b2, c_a = mk_block(b1, 2, c_a), c_a.evolve()
b3, c_a = mk_block(b2.id(), 3, c_a), c_a.evolve() b3, c_a = mk_block(b2, 3, c_a), c_a.evolve()
b4, c_b = mk_block(b1.id(), 2, c_b), c_b.evolve() b4, c_b = mk_block(b1, 2, c_b), c_b.evolve()
b5, c_b = mk_block(b4.id(), 3, c_b), c_b.evolve() b5, c_b = mk_block(b4, 3, c_b), c_b.evolve()
b6, c_c = mk_block(b4.id(), 3, c_c), c_c.evolve() b6, c_c = mk_block(b4, 3, c_c), c_c.evolve()
for b in [b1, b2, b3, b4, b5, b6]: for b in [b1, b2, b3, b4, b5, b6]:
follower.on_block(b) follower.on_block(b)
assert follower.tip() == b3 assert follower.tip() == b3
assert [f.tip() for f in follower.forks] == [b5, b6] assert [f for f in follower.forks] == [b5.id(), b6.id()]
assert follower.unimported_orphans(follower.tip_id()) == [b4, b5, b6] assert follower.unimported_orphans() == [b4, b5, b6]
b7, c_a = mk_block(b3.id(), 4, c_a, orphaned_proofs=[b4, b5, b6]), c_a.evolve() b7, c_a = mk_block(b3, 4, c_a, orphaned_proofs=[b4, b5, b6]), c_a.evolve()
follower.on_block(b7) follower.on_block(b7)
assert follower.tip() == b7 assert follower.tip() == b7
@ -235,30 +222,30 @@ class TestOrphanedProofs(TestCase):
follower = Follower(genesis, config) follower = Follower(genesis, config)
b1, c_a = mk_block(genesis.block, 1, c_a), c_a.evolve() b1, c_a = mk_block(genesis.block, 1, c_a), c_a.evolve()
b2, c_a = mk_block(b1.id(), 2, c_a), c_a.evolve() b2, c_a = mk_block(b1, 2, c_a), c_a.evolve()
b3, c_a = mk_block(b2.id(), 3, c_a), c_a.evolve() b3, c_a = mk_block(b2, 3, c_a), c_a.evolve()
b4, c_b = mk_block(b3.id(), 4, c_b), c_b.evolve() b4, c_b = mk_block(b3, 4, c_b), c_b.evolve()
b5, c_a = mk_block(b3.id(), 4, c_a), c_a.evolve() b5, c_a = mk_block(b3, 4, c_a), c_a.evolve()
b6, c_b = mk_block(b4.id(), 5, c_b, orphaned_proofs=[b5]), c_b.evolve() b6, c_b = mk_block(b4, 5, c_b, orphaned_proofs=[b5]), c_b.evolve()
b7, c_a = mk_block(b4.id(), 5, c_a, orphaned_proofs=[b5]), c_a.evolve() b7, c_a = mk_block(b4, 5, c_a, orphaned_proofs=[b5]), c_a.evolve()
b8, c_b = mk_block(b6.id(), 6, c_b, orphaned_proofs=[b7]), c_b.evolve() b8, c_b = mk_block(b6, 6, c_b, orphaned_proofs=[b7]), c_b.evolve()
for b in [b1, b2, b3, b4, b5]: for b in [b1, b2, b3, b4, b5]:
follower.on_block(b) follower.on_block(b)
assert follower.tip() == b4 assert follower.tip() == b4
assert follower.unimported_orphans(follower.tip_id()) == [b5] assert follower.unimported_orphans() == [b5]
for b in [b6, b7]: for b in [b6, b7]:
follower.on_block(b) follower.on_block(b)
assert follower.tip() == b6 assert follower.tip() == b6
assert follower.unimported_orphans(follower.tip_id()) == [b7] assert follower.unimported_orphans() == [b7]
follower.on_block(b8) follower.on_block(b8)
assert follower.tip() == b8 assert follower.tip() == b8
assert follower.unimported_orphans(follower.tip_id()) == [] assert follower.unimported_orphans() == []

7
cryptarchia/test_stake_relativization.py
View File

@ -1,5 +1,4 @@
from unittest import TestCase from unittest import TestCase
from dataclasses import dataclass
import itertools import itertools
import numpy as np import numpy as np
@ -29,13 +28,13 @@ class TestStakeRelativization(TestCase):
# on fork, tip state is not updated # on fork, tip state is not updated
orphan = mk_block(genesis.block, slot=1, coin=c_b) orphan = mk_block(genesis.block, slot=1, coin=c_b)
follower.on_block(orphan) follower.on_block(orphan)
assert follower.tip_state().block == b1.id() assert follower.tip_state().block == b1
assert follower.tip_state().leader_count == 1 assert follower.tip_state().leader_count == 1
# after orphan is adopted, leader count should jumpy by 2 (each orphan counts as a leader) # after orphan is adopted, leader count should jumpy by 2 (each orphan counts as a leader)
b2 = mk_block(b1.id(), slot=2, coin=c_a.evolve(), orphaned_proofs=[orphan]) b2 = mk_block(b1, slot=2, coin=c_a.evolve(), orphaned_proofs=[orphan])
follower.on_block(b2) follower.on_block(b2)
assert follower.tip_state().block == b2.id() assert follower.tip_state().block == b2
assert follower.tip_state().leader_count == 3 assert follower.tip_state().leader_count == 3
def test_inference_on_empty_genesis_epoch(self): def test_inference_on_empty_genesis_epoch(self):