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

Cryptarchia/drop orphan proofs (#121)

Browse Source 
* cryptarchia: introduce Hash class

* cryptarchia: Coin renamed to Note

* cryptarchia: simplify mock leader proof

* cryptarchia: remove orphan proofs from block headers

* cryptarchia: maintain a single commitment set in ledger state

* cryptarchia: drop note evolution

* cryptarchia: drop MOCK_LEADER_VRF

* cryptarchia fix nonce contribution derivation

* cryptarchia: mk_chain only returns list now

* fixup

* cryptarchia: shorten test cases using mk_chain
This commit is contained in:
davidrusu 2025-03-21 20:49:04 +04:00 committed by GitHub
parent f4b68f33cd
commit dcdb419648
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
9 changed files with 361 additions and 864 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

383
cryptarchia/cryptarchia.py
View File

@ -13,7 +13,19 @@ import numpy as np
logger = logging.getLogger(__name__)
Id: TypeAlias = bytes
class Hash(bytes):
ORDER = 2**256
def __new__(cls, dst, *data):
assert isinstance(dst, bytes)
h = sha256()
h.update(dst)
for d in data:
h.update(d)
return super().__new__(cls, h.digest())
def __deepcopy__(self, memo):
return self
@dataclass(frozen=True)
@ -128,10 +140,16 @@ class Slot:
@dataclass
class Coin:
sk: int
class Note:
value: int
nonce: bytes = bytes(32)
sk: int # TODO: rename to nf_sk
nonce: Hash = Hash(b"nonce")
unit: Hash = Hash(b"NMO")
state: Hash = Hash(b"state")
zone_id: Hash = Hash(b"ZoneID")
def __post_init__(self):
assert 0 <= self.value <= 2**64
@property
def pk(self) -> int:
@ -143,112 +161,73 @@ class Coin:
def encode_pk(self) -> bytes:
return int.to_bytes(self.pk, length=32, byteorder="big")
def evolve(self) -> "Coin":
h = blake2b(digest_size=32)
h.update(b"coin-evolve")
h.update(self.encode_sk())
h.update(self.nonce)
evolved_nonce = h.digest()
return Coin(nonce=evolved_nonce, sk=self.sk, value=self.value)
def commitment(self) -> Id:
# TODO: mocked until CL is understood
def commitment(self) -> Hash:
value_bytes = int.to_bytes(self.value, length=32, byteorder="big")
return Hash(
b"NOMOS_NOTE_CM",
self.state,
value_bytes,
self.unit,
self.nonce,
self.encode_pk(),
self.zone_id,
)
h = sha256()
h.update(b"coin-commitment")
h.update(self.nonce)
h.update(self.encode_pk())
h.update(value_bytes)
return h.digest()
def nullifier(self) -> Id:
# TODO: mocked until CL is understood
value_bytes = int.to_bytes(self.value, length=32, byteorder="big")
h = sha256()
h.update(b"coin-nullifier")
h.update(self.nonce)
h.update(self.encode_pk())
h.update(value_bytes)
return h.digest()
def nullifier(self) -> Hash:
return Hash(b"NOMOS_NOTE_NF", self.commitment(), self.encode_sk())
@dataclass
class MockLeaderProof:
commitment: Id
nullifier: Id
evolved_commitment: Id
note: Note
slot: Slot
parent: Id
parent: Hash
@staticmethod
def new(coin: Coin, slot: Slot, parent: Id):
evolved_coin = coin.evolve()
return MockLeaderProof(
commitment=coin.commitment(),
nullifier=coin.nullifier(),
evolved_commitment=evolved_coin.commitment(),
slot=slot,
parent=parent,
def epoch_nonce_contribution(self) -> Hash:
return Hash(
b"NOMOS_NONCE_CONTRIB",
self.slot.encode(),
self.note.commitment(),
self.note.encode_sk(),
)
def verify(self, slot: Slot, parent: Id):
# TODO: verification not implemented
return slot == self.slot and parent == self.parent
def verify(
self, slot: Slot, parent: Hash, commitments: set[Hash], nullifiers: set[Hash]
):
# TODO: verify slot lottery
return (
slot == self.slot
and parent == self.parent
and self.note.commitment() in commitments
and self.note.nullifier() not in nullifiers
)
@dataclass
class BlockHeader:
slot: Slot
parent: Id
parent: Hash
content_size: int
content_id: Id
content_id: Hash
leader_proof: MockLeaderProof
orphaned_proofs: List["BlockHeader"] = field(default_factory=list)
def update_header_hash(self, h):
# version byte
h.update(b"\x01")
# content size
h.update(int.to_bytes(self.content_size, length=4, byteorder="big"))
# content id
assert len(self.content_id) == 32
h.update(self.content_id)
# slot
h.update(self.slot.encode())
# parent
assert len(self.parent) == 32
h.update(self.parent)
# leader proof
assert len(self.leader_proof.commitment) == 32
h.update(self.leader_proof.commitment)
assert len(self.leader_proof.nullifier) == 32
h.update(self.leader_proof.nullifier)
assert len(self.leader_proof.evolved_commitment) == 32
h.update(self.leader_proof.evolved_commitment)
# orphaned proofs
h.update(int.to_bytes(len(self.orphaned_proofs), length=4, byteorder="big"))
for proof in self.orphaned_proofs:
proof.update_header_hash(h)
# **Attention**:
# The ID of a block header is defined as the 32byte blake2b hash of its fields
# The ID of a block header is defined as the hash of its fields
# as serialized in the format specified by the 'HEADER' rule in 'messages.abnf'.
#
# The following code is to be considered as a reference implementation, mostly to be used for testing.
def id(self) -> Id:
h = blake2b(digest_size=32)
self.update_header_hash(h)
return h.digest()
def id(self) -> Hash:
return Hash(
b"BLOCK_ID",
b"\x01", # version
int.to_bytes(self.content_size, length=4, byteorder="big"), # content size
self.content_id, # content id
self.slot.encode(), # slot
self.parent, # parent
# leader proof
self.leader_proof.epoch_nonce_contribution(),
# self.leader_proof -- the proof itself needs to be include in the hash
)
def __hash__(self):
return hash(self.id())
@ -264,23 +243,17 @@ class LedgerState:
# 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
# leader proof's nullifier.
#
# NOTE that this does not prevent nonce grinding at the last slot
# when the nonce snapshot is taken
nonce: Id = None
# leader proof's nonce contribution
nonce: Hash = None
# set of commitments
commitments_spend: set[Id] = field(default_factory=set)
# set of note commitments
commitments: set[Hash] = field(default_factory=set)
# set of commitments eligible to lead
commitments_lead: set[Id] = field(default_factory=set)
# set of nullified coins
nullifiers: set[Id] = field(default_factory=set)
# set of nullified notes
nullifiers: set[Hash] = field(default_factory=set)
# -- Stake Relativization State
# The number of observed leaders (blocks + orphans), this measurement is
# The number of observed leaders, this measurement is
# used in inferring total active stake in the network.
leader_count: int = 0
@ -288,8 +261,7 @@ class LedgerState:
return LedgerState(
block=self.block,
nonce=self.nonce,
commitments_spend=deepcopy(self.commitments_spend),
commitments_lead=deepcopy(self.commitments_lead),
commitments=deepcopy(self.commitments),
nullifiers=deepcopy(self.nullifiers),
leader_count=self.leader_count,
)
@ -297,34 +269,17 @@ class LedgerState:
def replace(self, **kwarg) -> "LedgerState":
return replace(self, **kwarg)
def verify_eligible_to_spend(self, commitment: Id) -> bool:
return commitment in self.commitments_spend
def verify_eligible_to_lead(self, commitment: Id) -> bool:
return commitment in self.commitments_lead
def verify_unspent(self, nullifier: Id) -> bool:
return nullifier not in self.nullifiers
def apply(self, block: BlockHeader):
assert block.parent == self.block.id()
h = blake2b(digest_size=32)
h.update("epoch-nonce".encode(encoding="utf-8"))
h.update(self.nonce)
h.update(block.leader_proof.nullifier)
h.update(block.slot.encode())
self.nonce = h.digest()
self.block = block
for proof in itertools.chain(block.orphaned_proofs, [block]):
self.apply_leader_proof(proof.leader_proof)
def apply_leader_proof(self, proof: MockLeaderProof):
self.nullifiers.add(proof.nullifier)
self.commitments_spend.add(proof.evolved_commitment)
self.commitments_lead.add(proof.evolved_commitment)
self.nonce = Hash(
b"EPOCH_NONCE",
self.nonce,
block.leader_proof.epoch_nonce_contribution(),
block.slot.encode(),
)
self.leader_count += 1
self.block = block
@dataclass
@ -343,20 +298,10 @@ class EpochState:
# leadership lottery.
inferred_total_active_stake: int
def verify_eligible_to_lead_due_to_age(self, commitment: Id) -> bool:
# A coin is eligible to lead if it was committed to before the the stake
# distribution snapshot was taken or it was produced by a leader proof
# since the snapshot was taken.
#
# This verification is checking that first condition.
#
# NOTE: `ledger_state.commitments_spend` is a super-set of `ledger_state.commitments_lead`
return self.stake_distribution_snapshot.verify_eligible_to_spend(commitment)
def total_active_stake(self) -> int:
"""
Returns the inferred total stake participating in consensus.
Total active stake is used to reletivize a coin's value in leadership proofs.
Total active stake is used to reletivize a note's value in leadership proofs.
"""
return self.inferred_total_active_stake
@ -367,7 +312,7 @@ class EpochState:
class Follower:
def __init__(self, genesis_state: LedgerState, config: Config):
self.config = config
self.forks: list[Id] = []
self.forks: list[Hash] = []
self.local_chain = genesis_state.block.id()
self.genesis_state = genesis_state
self.ledger_state = {genesis_state.block.id(): genesis_state.copy()}
@ -380,70 +325,19 @@ class Follower:
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
for orphan in block.orphaned_proofs:
# orphan proofs are checked in two ways
# 1. ensure they are valid locally in their original branch
# 2. ensure it does not conflict with current state
# We take a shortcut for (1.) by restricting orphans to proofs we've
# already processed in other branches.
if orphan.id() not in self.ledger_state:
raise MissingOrphanProof
# (2.) is satisfied by verifying the proof against current state ensuring:
# - it is a valid proof
# - and the nullifier has not already been spent
if not self.verify_slot_leader(
orphan.slot,
orphan.parent,
orphan.leader_proof,
epoch_state,
current_state,
):
raise InvalidOrphanProof
# if an adopted leadership proof is valid we need to apply its
# effects to the ledger state
current_state.apply_leader_proof(orphan.leader_proof)
# TODO: this is not the full block validation spec, only slot leader is verified
if not self.verify_slot_leader(
if not block.leader_proof.verify(
block.slot,
block.parent,
block.leader_proof,
epoch_state,
current_state,
epoch_state.stake_distribution_snapshot.commitments,
current_state.nullifiers,
):
raise InvalidLeaderProof
def verify_slot_leader(
self,
slot: Slot,
parent: Id,
proof: MockLeaderProof,
# coins are old enough if their commitment is in the stake distribution snapshot
epoch_state: EpochState,
# nullifiers (and commitments) are checked against the current state.
# For now, we assume proof parent state and current state are identical.
# This will change once we start putting merkle roots in headers
current_state: LedgerState,
) -> bool:
return (
proof.verify(slot, parent) # verify slot leader proof
and (
current_state.verify_eligible_to_lead(proof.commitment)
or epoch_state.verify_eligible_to_lead_due_to_age(proof.commitment)
)
and current_state.verify_unspent(proof.nullifier)
)
def apply_block_to_ledger_state(self, block: BlockHeader) -> bool:
if block.id() in self.ledger_state:
logger.warning("dropping already processed block")
@ -487,29 +381,8 @@ class Follower:
self.forks.remove(checkpoint_block_id)
self.local_chain = checkpoint_block_id
def unimported_orphans(self) -> list[BlockHeader]:
"""
Returns all unimported orphans w.r.t. the given tip's state.
Orphans are returned in the order that they should be imported.
"""
tip_state = self.tip_state().copy()
tip = tip_state.block.id()
orphans = []
for fork in self.forks:
_, _, fork_depth, fork_suffix = common_prefix_depth(
tip, fork, self.ledger_state
)
for b in fork_suffix:
if b.leader_proof.nullifier not in tip_state.nullifiers:
tip_state.nullifiers.add(b.leader_proof.nullifier)
orphans += [b]
return orphans
# Evaluate the fork choice rule and return the chain we should be following
def fork_choice(self) -> Id:
def fork_choice(self) -> Hash:
return maxvalid_bg(
self.local_chain,
self.forks,
@ -521,13 +394,13 @@ class Follower:
def tip(self) -> BlockHeader:
return self.tip_state().block
def tip_id(self) -> Id:
def tip_id(self) -> Hash:
return self.local_chain
def tip_state(self) -> LedgerState:
return self.ledger_state[self.tip_id()]
def state_at_slot_beginning(self, tip: Id, slot: Slot) -> LedgerState:
def state_at_slot_beginning(self, tip: Hash, slot: Slot) -> LedgerState:
for state in iter_chain(tip, self.ledger_state):
if state.block.slot < slot:
return state
@ -536,7 +409,7 @@ class Follower:
def epoch_start_slot(self, epoch) -> Slot:
return Slot(epoch.epoch * self.config.epoch_length)
def stake_distribution_snapshot(self, epoch, tip: Id):
def stake_distribution_snapshot(self, epoch, tip: Hash):
# 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
slot = Slot(epoch.prev().epoch * self.config.epoch_length)
@ -551,7 +424,7 @@ class Follower:
)
return self.state_at_slot_beginning(tip, slot)
def compute_epoch_state(self, epoch: Epoch, tip: Id) -> EpochState:
def compute_epoch_state(self, epoch: Epoch, tip: Hash) -> EpochState:
if epoch.epoch == 0:
return EpochState(
stake_distribution_snapshot=self.genesis_state,
@ -632,50 +505,34 @@ def phi(f: float, alpha: float) -> float:
return 1 - (1 - f) ** alpha
class MOCK_LEADER_VRF:
"""NOT SECURE: A mock VRF function"""
ORDER = 2**256
@classmethod
def vrf(cls, coin: Coin, epoch_nonce: bytes, slot: Slot) -> int:
h = sha256()
h.update(b"lead")
h.update(epoch_nonce)
h.update(slot.encode())
h.update(coin.encode_sk())
h.update(coin.nonce)
return int.from_bytes(h.digest())
@classmethod
def verify(cls, r, pk, nonce, slot):
raise NotImplemented()
@dataclass
class Leader:
config: Config
coin: Coin
note: Note
def try_prove_slot_leader(
self, epoch: EpochState, slot: Slot, parent: Id
self, epoch: EpochState, slot: Slot, parent: Hash
) -> MockLeaderProof | None:
if self._is_slot_leader(epoch, slot):
return MockLeaderProof.new(self.coin, slot, parent)
return MockLeaderProof(self.note, slot, parent)
def _is_slot_leader(self, epoch: EpochState, slot: Slot):
relative_stake = self.coin.value / epoch.total_active_stake()
relative_stake = self.note.value / epoch.total_active_stake()
r = MOCK_LEADER_VRF.vrf(self.coin, epoch.nonce(), slot)
return r < MOCK_LEADER_VRF.ORDER * phi(
self.config.active_slot_coeff, relative_stake
ticket = Hash(
b"LEAD",
epoch.nonce(),
slot.encode(),
self.note.commitment(),
self.note.encode_sk(),
)
ticket = int.from_bytes(ticket)
return ticket < Hash.ORDER * phi(self.config.active_slot_coeff, relative_stake)
def iter_chain(
tip: Id, states: Dict[Id, LedgerState]
tip: Hash, states: Dict[Hash, LedgerState]
) -> Generator[LedgerState, None, None]:
while tip in states:
yield states[tip]
@ -683,14 +540,14 @@ def iter_chain(
def iter_chain_blocks(
tip: Id, states: Dict[Id, LedgerState]
tip: Hash, states: Dict[Hash, LedgerState]
) -> Generator[BlockHeader, None, None]:
for state in iter_chain(tip, states):
yield state.block
def common_prefix_depth(
a: Id, b: Id, states: Dict[Id, LedgerState]
a: Hash, b: Hash, states: Dict[Hash, LedgerState]
) -> tuple[int, list[BlockHeader], int, list[BlockHeader]]:
return common_prefix_depth_from_chains(
iter_chain_blocks(a, states), iter_chain_blocks(b, states)
@ -748,7 +605,7 @@ def chain_density(chain: list[BlockHeader], slot: Slot) -> int:
return sum(1 for b in chain if b.slot < slot)
def block_children(states: Dict[Id, LedgerState]) -> Dict[Id, set[Id]]:
def block_children(states: Dict[Hash, LedgerState]) -> Dict[Hash, set[Hash]]:
children = defaultdict(set)
for c, state in states.items():
children[state.block.parent].add(c)
@ -764,14 +621,14 @@ def block_children(states: Dict[Id, LedgerState]) -> Dict[Id, set[Id]]:
# 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
def maxvalid_bg(
local_chain: Id,
forks: List[Id],
local_chain: Hash,
forks: List[Hash],
k: int,
s: int,
states: Dict[Id, LedgerState],
) -> Id:
assert type(local_chain) == Id
assert all(type(f) == Id for f in forks)
states: Dict[Hash, LedgerState],
) -> Hash:
assert type(local_chain) == Hash, type(local_chain)
assert all(type(f) == Hash for f in forks)
cmax = local_chain
for fork in forks:
@ -802,16 +659,6 @@ class ParentNotFound(Exception):
return "Parent not found"
class MissingOrphanProof(Exception):
def __str__(self):
return "Missing orphan proof"
class InvalidOrphanProof(Exception):
def __str__(self):
return "Invalid orphan proof"
class InvalidLeaderProof(Exception):
def __str__(self):
return "Invalid leader proof"

6
cryptarchia/sync.py
View File

@ -4,7 +4,7 @@ from typing import Generator
from cryptarchia.cryptarchia import (
BlockHeader,
Follower,
Id,
Hash,
ParentNotFound,
Slot,
common_prefix_depth_from_chains,
@ -19,7 +19,7 @@ def sync(local: Follower, peers: list[Follower]):
# Repeat the sync process until no peer has a tip ahead of the local tip,
# because peers' tips may advance during the sync process.
block_fetcher = BlockFetcher(peers)
rejected_blocks: set[Id] = set()
rejected_blocks: set[Hash] = set()
while True:
# Fetch blocks from the peers in the range of slots from the local tip to the latest tip.
# Gather orphaned blocks, which are blocks from forks that are absent in the local block tree.
@ -145,7 +145,7 @@ class BlockFetcher:
continue
def fetch_chain_backward(
self, tip: Id, local: Follower
self, tip: Hash, local: Follower
) -> Generator[BlockHeader, None, None]:
# Fetches a chain of blocks from the peers, starting from the given tip to the genesis.
# Attempts to extend the chain as much as possible by querying multiple peers,

39
cryptarchia/test_common.py
View File

@ -1,19 +1,20 @@
from .cryptarchia import (
Config,
Slot,
Coin,
Note,
BlockHeader,
LedgerState,
MockLeaderProof,
Leader,
Follower,
Hash,
)
class TestNode:
def __init__(self, config: Config, genesis: LedgerState, coin: Coin):
def __init__(self, config: Config, genesis: LedgerState, note: Note):
self.config = config
self.leader = Leader(coin=coin, config=config)
self.leader = Leader(note=note, config=config)
self.follower = Follower(genesis, config)
def epoch_state(self, slot: Slot):
@ -25,11 +26,9 @@ class TestNode:
parent = self.follower.tip_id()
epoch_state = self.epoch_state(slot)
if leader_proof := self.leader.try_prove_slot_leader(epoch_state, slot, parent):
self.leader.coin = self.leader.coin.evolve()
return BlockHeader(
parent=parent,
slot=slot,
orphaned_proofs=self.follower.unimported_orphans(),
leader_proof=leader_proof,
content_size=0,
content_id=bytes(32),
@ -40,57 +39,51 @@ class TestNode:
self.follower.on_block(block)
def mk_config(initial_stake_distribution: list[Coin]) -> Config:
initial_inferred_total_stake = sum(c.value for c in initial_stake_distribution)
def mk_config(initial_stake_distribution: list[Note]) -> Config:
initial_inferred_total_stake = sum(n.value for n in initial_stake_distribution)
return Config.cryptarchia_v0_0_1(initial_inferred_total_stake).replace(
k=1,
active_slot_coeff=0.5,
)
def mk_genesis_state(initial_stake_distribution: list[Coin]) -> LedgerState:
def mk_genesis_state(initial_stake_distribution: list[Note]) -> LedgerState:
return LedgerState(
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)
leader_proof=MockLeaderProof(
Note(sk=0, value=0), Slot(0), parent=bytes(32)
),
),
nonce=bytes(32),
commitments_spend={c.commitment() for c in initial_stake_distribution},
commitments_lead={c.commitment() for c in initial_stake_distribution},
commitments={n.commitment() for n in initial_stake_distribution},
nullifiers=set(),
)
def mk_block(
parent: BlockHeader, slot: int, coin: Coin, content=bytes(32), orphaned_proofs=[]
parent: BlockHeader, slot: int, note: Note, content=bytes(32)
) -> BlockHeader:
assert type(parent) == BlockHeader, type(parent)
assert type(slot) == int, type(slot)
from hashlib import sha256
return BlockHeader(
slot=Slot(slot),
parent=parent.id(),
content_size=len(content),
content_id=sha256(content).digest(),
leader_proof=MockLeaderProof.new(coin, Slot(slot), parent=parent.id()),
orphaned_proofs=orphaned_proofs,
content_id=Hash(b"CONTENT_ID", content),
leader_proof=MockLeaderProof(note, Slot(slot), parent=parent.id()),
)
def mk_chain(
parent: BlockHeader, coin: Coin, slots: list[int]
) -> tuple[list[BlockHeader], Coin]:
def mk_chain(parent: BlockHeader, note: Note, slots: list[int]) -> list[BlockHeader]:
assert type(parent) == BlockHeader
chain = []
for s in slots:
block = mk_block(parent=parent, slot=s, coin=coin)
block = mk_block(parent=parent, slot=s, note=note)
chain.append(block)
parent = block
coin = coin.evolve()
return chain, coin
return chain

60
cryptarchia/test_fork_choice.py
View File

@ -4,7 +4,7 @@ from copy import deepcopy
from cryptarchia.cryptarchia import (
maxvalid_bg,
Slot,
Coin,
Note,
Follower,
common_prefix_depth,
LedgerState,
@ -21,16 +21,12 @@ class TestForkChoice(TestCase):
# \
# 4 - 5
coin = Coin(sk=1, value=100)
note = Note(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)
b1, b2, b3 = mk_chain(b0, Note(sk=1, value=1), slots=[1, 2, 3])
b4, b5 = mk_chain(b0, Note(sk=2, value=1), slots=[1, 2])
b6, b7 = mk_chain(b2, Note(sk=3, value=1), slots=[3, 4])
states = {
b.id(): LedgerState(block=b) for b in [b0, b1, b2, b3, b4, b5, b6, b7]
@ -173,20 +169,20 @@ class TestForkChoice(TestCase):
# 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)
common, long_coin = mk_chain(parent=genesis, coin=long_coin, slots=range(50))
short_note, long_note = Note(sk=0, value=100), Note(sk=1, value=100)
common = mk_chain(parent=genesis, note=long_note, slots=range(50))
long_chain_sparse_ext, long_coin = mk_chain(
parent=common[-1], coin=long_coin, slots=range(50, 100, 2)
long_chain_sparse_ext = mk_chain(
parent=common[-1], note=long_note, slots=range(50, 100, 2)
)
short_chain_dense_ext, _ = mk_chain(
parent=common[-1], coin=short_coin, slots=range(50, 100)
short_chain_dense_ext = mk_chain(
parent=common[-1], note=short_note, slots=range(50, 100)
)
# add more blocks to the long chain to ensure the long chain is indeed longer
long_chain_further_ext, _ = mk_chain(
parent=long_chain_sparse_ext[-1], coin=long_coin, slots=range(100, 126)
long_chain_further_ext = mk_chain(
parent=long_chain_sparse_ext[-1], note=long_note, slots=range(100, 126)
)
long_chain = deepcopy(common) + long_chain_sparse_ext + long_chain_further_ext
@ -213,18 +209,18 @@ class TestForkChoice(TestCase):
def test_fork_choice_long_dense_chain(self):
# The longest chain is also the densest after the fork
short_coin, long_coin = Coin(sk=0, value=100), Coin(sk=1, value=100)
common, long_coin = mk_chain(
short_note, long_note = Note(sk=0, value=100), Note(sk=1, value=100)
common = mk_chain(
parent=mk_genesis_state([]).block,
coin=long_coin,
note=long_note,
slots=range(1, 50),
)
long_chain_dense_ext, _ = mk_chain(
parent=common[-1], coin=long_coin, slots=range(50, 100)
long_chain_dense_ext = mk_chain(
parent=common[-1], note=long_note, slots=range(50, 100)
)
short_chain_sparse_ext, _ = mk_chain(
parent=common[-1], coin=short_coin, slots=range(50, 100, 2)
short_chain_sparse_ext = mk_chain(
parent=common[-1], note=short_note, slots=range(50, 100, 2)
)
long_chain = deepcopy(common) + long_chain_dense_ext
@ -240,13 +236,13 @@ class TestForkChoice(TestCase):
)
def test_fork_choice_integration(self):
c_a, c_b = Coin(sk=0, value=10), Coin(sk=1, value=10)
coins = [c_a, c_b]
config = mk_config(coins)
genesis = mk_genesis_state(coins)
n_a, n_b = Note(sk=0, value=10), Note(sk=1, value=10)
notes = [n_a, n_b]
config = mk_config(notes)
genesis = mk_genesis_state(notes)
follower = Follower(genesis, config)
b1, c_a = mk_block(genesis.block, 1, c_a), c_a.evolve()
b1 = mk_block(genesis.block, 1, n_a)
follower.on_block(b1)
@ -262,8 +258,8 @@ class TestForkChoice(TestCase):
# b3
#
b2, c_a = mk_block(b1, 2, c_a), c_a.evolve()
b3, c_b = mk_block(b1, 2, c_b), c_b.evolve()
b2 = mk_block(b1, 2, n_a)
b3 = mk_block(b1, 2, n_b)
follower.on_block(b2)
follower.on_block(b3)
@ -280,7 +276,7 @@ class TestForkChoice(TestCase):
# b3 - b4 == tip
#
b4, c_b = mk_block(b3, 3, c_b), c_a.evolve()
b4 = mk_block(b3, 3, n_b)
follower.on_block(b4)
assert follower.tip_id() == b4.id()

8
cryptarchia/test_leader.py
View File

@ -2,7 +2,7 @@ from unittest import TestCase
import numpy as np
from .cryptarchia import Leader, EpochState, LedgerState, Coin, phi, Slot
from .cryptarchia import Leader, EpochState, LedgerState, Note, phi, Slot
from .test_common import mk_config
@ -14,11 +14,11 @@ class TestLeader(TestCase):
inferred_total_active_stake=1000,
)
coin = Coin(sk=0, value=10)
note = Note(sk=0, value=10)
f = 0.05
l = Leader(
config=mk_config([coin]).replace(active_slot_coeff=f),
coin=coin,
config=mk_config([note]).replace(active_slot_coeff=f),
note=note,
)
# We'll use the Margin of Error equation to decide how many samples we need.

223
cryptarchia/test_ledger_state_update.py
View File

@ -1,10 +1,9 @@
from unittest import TestCase
from .cryptarchia import (
Coin,
Note,
Follower,
InvalidLeaderProof,
MissingOrphanProof,
ParentNotFound,
iter_chain,
)
@ -13,12 +12,12 @@ from .test_common import mk_block, mk_config, mk_genesis_state
class TestLedgerStateUpdate(TestCase):
def test_on_block_idempotent(self):
leader_coin = Coin(sk=0, value=100)
genesis = mk_genesis_state([leader_coin])
leader_note = Note(sk=0, value=100)
genesis = mk_genesis_state([leader_note])
follower = Follower(genesis, mk_config([leader_coin]))
follower = Follower(genesis, mk_config([leader_note]))
block = mk_block(slot=0, parent=genesis.block, coin=leader_coin)
block = mk_block(slot=0, parent=genesis.block, note=leader_note)
follower.on_block(block)
# Follower should have accepted the block
@ -33,47 +32,42 @@ class TestLedgerStateUpdate(TestCase):
assert len(follower.ledger_state) == 2
assert len(follower.forks) == 0
def test_ledger_state_prevents_coin_reuse(self):
leader_coin = Coin(sk=0, value=100)
genesis = mk_genesis_state([leader_coin])
def test_ledger_state_allows_note_reuse(self):
leader_note = Note(sk=0, value=100)
genesis = mk_genesis_state([leader_note])
follower = Follower(genesis, mk_config([leader_coin]))
follower = Follower(genesis, mk_config([leader_note]))
block = mk_block(slot=0, parent=genesis.block, coin=leader_coin)
block = mk_block(slot=0, parent=genesis.block, note=leader_note)
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 should have updated their ledger state to mark the leader coin as spent
assert follower.tip_state().verify_unspent(leader_coin.nullifier()) == False
reuse_note_block = mk_block(slot=1, parent=block, note=leader_note)
follower.on_block(reuse_note_block)
reuse_coin_block = mk_block(slot=1, parent=block, coin=leader_coin)
with self.assertRaises(InvalidLeaderProof):
follower.on_block(reuse_coin_block)
# Follower should *not* have accepted the block
assert len(list(iter_chain(follower.tip_id(), follower.ledger_state))) == 2
assert follower.tip() == block
# Follower should have accepted the block
assert len(list(iter_chain(follower.tip_id(), follower.ledger_state))) == 3
assert follower.tip() == reuse_note_block
def test_ledger_state_is_properly_updated_on_reorg(self):
coin = [Coin(sk=0, value=100), Coin(sk=1, value=100), Coin(sk=2, value=100)]
note = [Note(sk=0, value=100), Note(sk=1, value=100), Note(sk=2, value=100)]
genesis = mk_genesis_state(coin)
genesis = mk_genesis_state(note)
follower = Follower(genesis, mk_config(coin))
follower = Follower(genesis, mk_config(note))
# 1) coin[0] & coin[1] both concurrently win slot 0
# 1) note[0] & note[1] both concurrently win slot 0
block_1 = mk_block(parent=genesis.block, slot=0, coin=coin[0])
block_2 = mk_block(parent=genesis.block, slot=0, coin=coin[1])
block_1 = mk_block(parent=genesis.block, slot=0, note=note[0])
block_2 = mk_block(parent=genesis.block, slot=0, note=note[1])
# 2) follower sees block 1 first
follower.on_block(block_1)
assert follower.tip() == block_1
assert not follower.tip_state().verify_unspent(coin[0].nullifier())
# 3) then sees block 2, but sticks with block_1 as the tip
@ -81,46 +75,43 @@ class TestLedgerStateUpdate(TestCase):
assert follower.tip() == block_1
assert len(follower.forks) == 1, f"{len(follower.forks)}"
# 4) then coin[2] wins slot 1 and chooses to extend from block_2
# 4) then note[2] wins slot 1 and chooses to extend from block_2
block_3 = mk_block(parent=block_2, slot=1, coin=coin[2])
block_3 = mk_block(parent=block_2, slot=1, note=note[2])
follower.on_block(block_3)
# the follower should have switched over to the block_2 fork
assert follower.tip() == block_3
# and the original coin[0] should now be removed from the spent pool
assert follower.tip_state().verify_unspent(coin[0].nullifier())
def test_fork_creation(self):
coins = [Coin(sk=i, value=100) for i in range(7)]
genesis = mk_genesis_state(coins)
notes = [Note(sk=i, value=100) for i in range(7)]
genesis = mk_genesis_state(notes)
follower = Follower(genesis, mk_config(coins))
follower = Follower(genesis, mk_config(notes))
# coin_0 & coin_1 both concurrently win slot 0 based on the genesis block
# note_0 & note_1 both concurrently win slot 0 based on the genesis block
# Both blocks are accepted, and a fork is created "from the genesis block"
block_1 = mk_block(parent=genesis.block, slot=0, coin=coins[0])
block_2 = mk_block(parent=genesis.block, slot=0, coin=coins[1])
block_1 = mk_block(parent=genesis.block, slot=0, note=notes[0])
block_2 = mk_block(parent=genesis.block, slot=0, note=notes[1])
follower.on_block(block_1)
follower.on_block(block_2)
assert follower.tip() == block_1
assert len(follower.forks) == 1, f"{len(follower.forks)}"
assert follower.forks[0] == block_2.id()
# 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
# note_2 wins slot 1 and chooses to extend from block_1
# note_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.
block_3 = mk_block(parent=block_1, slot=1, coin=coins[2])
block_4 = mk_block(parent=block_2, slot=1, coin=coins[3])
block_3 = mk_block(parent=block_1, slot=1, note=notes[2])
block_4 = mk_block(parent=block_2, slot=1, note=notes[3])
follower.on_block(block_3)
follower.on_block(block_4)
assert follower.tip() == block_3
assert len(follower.forks) == 1, f"{len(follower.forks)}"
assert follower.forks[0] == block_4.id()
# coin_4 wins slot 1 and but chooses to extend from block_2 as well
# note_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".
block_5 = mk_block(parent=block_2, slot=1, coin=coins[4])
block_5 = mk_block(parent=block_2, slot=1, note=notes[4])
follower.on_block(block_5)
assert follower.tip() == block_3
assert len(follower.forks) == 2, f"{len(follower.forks)}"
@ -129,8 +120,8 @@ class TestLedgerStateUpdate(TestCase):
# A block based on an unknown parent is not accepted.
# Nothing changes from the local chain and forks.
unknown_block = mk_block(parent=block_5, slot=2, coin=coins[5])
block_6 = mk_block(parent=unknown_block, slot=2, coin=coins[6])
unknown_block = mk_block(parent=block_5, slot=2, note=notes[5])
block_6 = mk_block(parent=unknown_block, slot=2, note=notes[6])
with self.assertRaises(ParentNotFound):
follower.on_block(block_6)
assert follower.tip() == block_3
@ -139,9 +130,9 @@ class TestLedgerStateUpdate(TestCase):
assert follower.forks[1] == block_5.id()
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(leader_coins)
leader_notes = [Note(sk=i, value=100) for i in range(4)]
genesis = mk_genesis_state(leader_notes)
config = mk_config(leader_notes)
follower = Follower(genesis, config)
@ -150,19 +141,19 @@ class TestLedgerStateUpdate(TestCase):
# ---- EPOCH 0 ----
block_1 = mk_block(slot=0, parent=genesis.block, coin=leader_coins[0])
block_1 = mk_block(slot=0, parent=genesis.block, note=leader_notes[0])
follower.on_block(block_1)
assert follower.tip() == block_1
assert follower.tip().slot.epoch(config).epoch == 0
block_2 = mk_block(slot=19, parent=block_1, coin=leader_coins[1])
block_2 = mk_block(slot=19, parent=block_1, note=leader_notes[1])
follower.on_block(block_2)
assert follower.tip() == block_2
assert follower.tip().slot.epoch(config).epoch == 0
# ---- EPOCH 1 ----
block_3 = mk_block(slot=20, parent=block_2, coin=leader_coins[2])
block_3 = mk_block(slot=20, parent=block_2, note=leader_notes[2])
follower.on_block(block_3)
assert follower.tip() == block_3
assert follower.tip().slot.epoch(config).epoch == 1
@ -171,48 +162,51 @@ class TestLedgerStateUpdate(TestCase):
# 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,
# To ensure this is the case, we add a new note 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
block_4 = mk_block(slot=40, parent=block_3, coin=Coin(sk=4, value=100))
block_4 = mk_block(slot=40, parent=block_3, note=Note(sk=4, value=100))
with self.assertRaises(InvalidLeaderProof):
follower.on_block(block_4)
assert follower.tip() == block_3
# then we add the coin to "spendable commitments" associated with slot 9
follower.ledger_state[block_2.id()].commitments_spend.add(
Coin(sk=4, value=100).commitment()
# then we add the note to "commitments" associated with slot 9
follower.ledger_state[block_2.id()].commitments.add(
Note(sk=4, value=100).commitment()
)
follower.on_block(block_4)
assert follower.tip() == block_4
assert follower.tip().slot.epoch(config).epoch == 2
def test_evolved_coin_is_eligible_for_leadership(self):
coin = Coin(sk=0, value=100)
def test_note_added_after_stake_freeze_is_ineligible_for_leadership(self):
note = Note(sk=0, value=100)
genesis = mk_genesis_state([coin])
genesis = mk_genesis_state([note])
follower = Follower(genesis, mk_config([coin]))
follower = Follower(genesis, mk_config([note]))
# coin wins the first slot
block_1 = mk_block(slot=0, parent=genesis.block, coin=coin)
# note wins the first slot
block_1 = mk_block(slot=0, parent=genesis.block, note=note)
follower.on_block(block_1)
assert follower.tip() == block_1
# coin can't be reused to win following slots:
block_2_reuse = mk_block(slot=1, parent=block_1, coin=coin)
# note can be reused to win following slots:
block_2 = mk_block(slot=1, parent=block_1, note=note)
follower.on_block(block_2)
assert follower.tip() == block_2
# but the a new note is ineligible
note_new = Note(sk=1, value=10)
follower.tip_state().commitments.add(note_new.commitment())
block_3_new = mk_block(slot=2, parent=block_2, note=note_new)
with self.assertRaises(InvalidLeaderProof):
follower.on_block(block_2_reuse)
assert follower.tip() == block_1
follower.on_block(block_3_new)
# but the evolved coin is eligible
block_2_evolve = mk_block(slot=1, parent=block_1, coin=coin.evolve())
follower.on_block(block_2_evolve)
assert follower.tip() == block_2_evolve
assert follower.tip() == block_2
def test_new_coins_becoming_eligible_after_stake_distribution_stabilizes(self):
coin = Coin(sk=0, value=100)
config = mk_config([coin])
genesis = mk_genesis_state([coin])
def test_new_notes_becoming_eligible_after_stake_distribution_stabilizes(self):
note = Note(sk=0, value=100)
config = mk_config([note])
genesis = mk_genesis_state([note])
follower = Follower(genesis, config)
# We assume an epoch length of 20 slots in this test.
@ -220,89 +214,34 @@ class TestLedgerStateUpdate(TestCase):
# ---- EPOCH 0 ----
block_0_0 = mk_block(slot=0, parent=genesis.block, coin=coin)
block_0_0 = mk_block(slot=0, parent=genesis.block, note=note)
follower.on_block(block_0_0)
assert follower.tip() == block_0_0
# mint a new coin to be used for leader elections in upcoming epochs
coin_new = Coin(sk=1, value=10)
follower.ledger_state[block_0_0.id()].commitments_spend.add(
coin_new.commitment()
)
# mint a new note to be used for leader elections in upcoming epochs
note_new = Note(sk=1, value=10)
follower.ledger_state[block_0_0.id()].commitments.add(note_new.commitment())
# the new coin is not yet eligible for elections
block_0_1_attempt = mk_block(slot=1, parent=block_0_0, coin=coin_new)
# the new note is not yet eligible for elections
block_0_1_attempt = mk_block(slot=1, parent=block_0_0, note=note_new)
with self.assertRaises(InvalidLeaderProof):
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, coin=coin.evolve())
follower.on_block(block_0_1)
assert follower.tip() == block_0_1
# ---- EPOCH 1 ----
# The newly minted coin is still not eligible in the following epoch since the
# The newly minted note is still not eligible in the following epoch since the
# stake distribution snapshot is taken at the beginning of the previous epoch
block_1_0 = mk_block(slot=20, parent=block_0_1, coin=coin_new)
block_1_0_attempt = mk_block(slot=20, parent=block_0_0, note=note_new)
with self.assertRaises(InvalidLeaderProof):
follower.on_block(block_1_0)
assert follower.tip() == block_0_1
follower.on_block(block_1_0_attempt)
assert follower.tip() == block_0_0
# ---- EPOCH 2 ----
# The coin is finally eligible 2 epochs after it was first minted
# The note is finally eligible 2 epochs after it was first minted
block_2_0 = mk_block(slot=40, parent=block_0_1, coin=coin_new)
block_2_0 = mk_block(slot=40, parent=block_0_0, note=note_new)
follower.on_block(block_2_0)
assert follower.tip() == block_2_0
# 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, coin=coin_new.evolve())
follower.on_block(block_2_1)
assert follower.tip() == block_2_1
def test_orphaned_proofs(self):
coin, coin_orphan = Coin(sk=0, value=100), Coin(sk=1, value=100)
genesis = mk_genesis_state([coin, coin_orphan])
follower = Follower(genesis, mk_config([coin, coin_orphan]))
block_0_0 = mk_block(slot=0, parent=genesis.block, coin=coin)
follower.on_block(block_0_0)
assert follower.tip() == block_0_0
coin_new = coin.evolve()
coin_new_new = coin_new.evolve()
block_0_1 = mk_block(slot=1, parent=block_0_0, coin=coin_new_new)
with self.assertRaises(InvalidLeaderProof):
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 will not be accepted until a node first sees the corresponding block.
#
# Also, notice that the block is using the evolved orphan coin which is not present on the main
# branch. The evolved orphan commitment is added from the orphan prior to validating the block
# header as part of orphan importing process
orphan = mk_block(parent=genesis.block, slot=0, coin=coin_orphan)
block_0_1 = mk_block(
slot=1,
parent=block_0_0,
coin=coin_orphan.evolve(),
orphaned_proofs=[orphan],
)
with self.assertRaises(MissingOrphanProof):
follower.on_block(block_0_1)
# since follower had not seen this orphan prior to being included as
# an orphan proof, it will be rejected
assert follower.tip() == block_0_0
# but all is fine if the follower first sees the orphan block, and then
# is imported into the main chain
follower.on_block(orphan)
follower.on_block(block_0_1)
assert follower.tip() == block_0_1

251
cryptarchia/test_orphaned_proofs.py
View File

@ -1,251 +0,0 @@
from unittest import TestCase
from cryptarchia.cryptarchia import Coin, Follower
from .test_common import mk_config, mk_genesis_state, mk_block
class TestOrphanedProofs(TestCase):
def test_simple_orphan_import(self):
c_a, c_b = Coin(sk=0, value=10), Coin(sk=1, value=10)
coins = [c_a, c_b]
config = mk_config(coins)
genesis = mk_genesis_state(coins)
follower = Follower(genesis, config)
# -- fork --
#
# b2 == tip
# /
# b1
# \
# b3
#
b1, c_a = mk_block(genesis.block, 1, c_a), c_a.evolve()
b2, c_a = mk_block(b1, 2, c_a), c_a.evolve()
b3, c_b = mk_block(b1, 2, c_b), c_b.evolve()
for b in [b1, b2, b3]:
follower.on_block(b)
assert follower.tip() == b2
assert [f for f in follower.forks] == [b3.id()]
assert follower.unimported_orphans() == [b3]
# -- extend with import --
#
# b2 - b4
# / /
# b1 /
# \ /
# b3
#
b4, c_a = mk_block(b2, 3, c_a, orphaned_proofs=[b3]), c_a.evolve()
follower.on_block(b4)
assert follower.tip() == b4
assert [f for f in follower.forks] == [b3.id()]
assert follower.unimported_orphans() == []
def test_orphan_proof_import_from_long_running_fork(self):
c_a, c_b = Coin(sk=0, value=10), Coin(sk=1, value=10)
coins = [c_a, c_b]
config = mk_config(coins)
genesis = mk_genesis_state(coins)
follower = Follower(genesis, config)
# -- fork --
#
# b2 - b3 == tip
# /
# b1
# \
# b4 - b5
#
b1, c_a = mk_block(genesis.block, 1, c_a), c_a.evolve()
b2, c_a = mk_block(b1, 2, c_a), c_a.evolve()
b3, c_a = mk_block(b2, 3, c_a), c_a.evolve()
b4, c_b = mk_block(b1, 2, 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]:
follower.on_block(b)
assert follower.tip() == b3
assert [f for f in follower.forks] == [b5.id()]
assert follower.unimported_orphans() == [b4, b5]
# -- extend b3, importing the fork --
#
# b2 - b3 - b6 == tip
# / ___/
# b1 ___/ /
# \ / /
# b4 - b5
b6, c_a = mk_block(b3, 4, c_a, orphaned_proofs=[b4, b5]), c_a.evolve()
follower.on_block(b6)
assert follower.tip() == b6
assert [f for f in follower.forks] == [b5.id()]
def test_orphan_proof_import_from_fork_without_direct_shared_parent(self):
coins = [Coin(sk=i, value=10) for i in range(2)]
c_a, c_b = coins
config = mk_config(coins)
genesis = mk_genesis_state(coins)
follower = Follower(genesis, config)
# -- forks --
#
# b2 - b3 - b4 == tip
# /
# b1
# \
# b5 - b6 - b7
b1, c_a = mk_block(genesis.block, 1, c_a), c_a.evolve()
b2, c_a = mk_block(b1, 2, c_a), c_a.evolve()
b3, c_a = mk_block(b2, 3, c_a), c_a.evolve()
b4, c_a = mk_block(b3, 4, c_a), c_a.evolve()
b5, c_b = mk_block(b1, 2, c_b), c_b.evolve()
b6, c_b = mk_block(b5, 3, 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]:
follower.on_block(b)
assert follower.tip() == b4
assert [f for f in follower.forks] == [b7.id()]
assert follower.unimported_orphans() == [b5, b6, b7]
# -- extend b4, importing the forks --
#
# b2 - b3 - b4 - b8 == tip
# / _______/
# b1 ____/______/
# \ / / /
# b5 - b6 - b7
#
# Earlier implementations of orphan proof validation failed to
# validate b7 as an orphan here.
b8, c_a = mk_block(b4, 5, c_a, orphaned_proofs=[b5, b6, b7]), c_a.evolve()
follower.on_block(b8)
assert follower.tip() == b8
assert [f for f in follower.forks] == [b7.id()]
assert follower.unimported_orphans() == []
def test_unimported_orphans(self):
# Given the following fork graph:
#
# b2 - b3
# /
# b1
# \
# b4 - b5
# \
# -- b6
#
# Orphans w.r.t. to b3 are b4..6, thus extending from b3 with b7 would
# give the following fork graph
#
# b2 - b3 --- b7== tip
# / ____/
# b1 ____/ __/
# \ / / /
# b4 - b5 /
# \ /
# -- b6
#
coins = [Coin(sk=i, value=10) for i in range(3)]
c_a, c_b, c_c = coins
config = mk_config(coins)
genesis = mk_genesis_state(coins)
follower = Follower(genesis, config)
b1, c_a = mk_block(genesis.block, 1, c_a), c_a.evolve()
b2, c_a = mk_block(b1, 2, c_a), c_a.evolve()
b3, c_a = mk_block(b2, 3, c_a), c_a.evolve()
b4, c_b = mk_block(b1, 2, c_b), c_b.evolve()
b5, c_b = mk_block(b4, 3, c_b), c_b.evolve()
b6, c_c = mk_block(b4, 3, c_c), c_c.evolve()
for b in [b1, b2, b3, b4, b5, b6]:
follower.on_block(b)
assert follower.tip() == b3
assert [f for f in follower.forks] == [b5.id(), b6.id()]
assert follower.unimported_orphans() == [b4, b5, b6]
b7, c_a = mk_block(b3, 4, c_a, orphaned_proofs=[b4, b5, b6]), c_a.evolve()
follower.on_block(b7)
assert follower.tip() == b7
def test_transitive_orphan_reimports(self):
# Two forks, one after the other, with some complicated orphan imports.
# I don't have different line colors to differentiate orphans from parents
# so I've added o=XX to differentiate orphans from parents.
#
# - The first fork at b3(a) is not too interesting.
# - The second fork at b4(b) has both b6 and b7 importing b5
# - crucially b7 uses the evolved commitment from b5
# - Then finally b8 imports b7.
#
# proper orphan proof importing will be able to deal with the fact that
# b7's commitment was produced outside of the main branch AND the commitment
# is not part of the current list of orphans in b8
# (b5 had already been imported, therefore it is not included as an orphan in b8)
#
# b1(a) - b2(a) - b3(a) - b4(b) - b6(b, o=b5) - b8(b, o=b7)
# \ \___ __/ __/
# \ _x_ __/
# \ / \_ /
# -b5(a)-----\-b7(a, o=b5)
coins = [Coin(sk=i, value=10) for i in range(2)]
c_a, c_b = coins
config = mk_config(coins)
genesis = mk_genesis_state(coins)
follower = Follower(genesis, config)
b1, c_a = mk_block(genesis.block, 1, c_a), c_a.evolve()
b2, c_a = mk_block(b1, 2, c_a), c_a.evolve()
b3, c_a = mk_block(b2, 3, c_a), c_a.evolve()
b4, c_b = mk_block(b3, 4, c_b), c_b.evolve()
b5, c_a = mk_block(b3, 4, c_a), c_a.evolve()
b6, c_b = mk_block(b4, 5, c_b, orphaned_proofs=[b5]), c_b.evolve()
b7, c_a = mk_block(b4, 5, c_a, orphaned_proofs=[b5]), c_a.evolve()
b8, c_b = mk_block(b6, 6, c_b, orphaned_proofs=[b7]), c_b.evolve()
for b in [b1, b2, b3, b4, b5]:
follower.on_block(b)
assert follower.tip() == b4
assert follower.unimported_orphans() == [b5]
for b in [b6, b7]:
follower.on_block(b)
assert follower.tip() == b6
assert follower.unimported_orphans() == [b7]
follower.on_block(b8)
assert follower.tip() == b8
assert follower.unimported_orphans() == []

36
cryptarchia/test_stake_relativization.py
View File

@ -3,17 +3,17 @@ import itertools
import numpy as np
from .cryptarchia import Config, Coin, Slot
from .cryptarchia import Config, Note, Slot
from .test_common import mk_config, mk_genesis_state, mk_block, TestNode, Follower
class TestStakeRelativization(TestCase):
def test_ledger_leader_counting(self):
coins = [Coin(sk=i, value=10) for i in range(2)]
c_a, c_b = coins
notes = [Note(sk=i, value=10) for i in range(2)]
n_a, n_b = notes
config = mk_config(coins)
genesis = mk_genesis_state(coins)
config = mk_config(notes)
genesis = mk_genesis_state(notes)
follower = Follower(genesis, config)
@ -21,31 +21,31 @@ class TestStakeRelativization(TestCase):
assert follower.tip_state().leader_count == 0
# after a block, 1 leader has been observed
b1 = mk_block(genesis.block, slot=1, coin=c_a)
b1 = mk_block(genesis.block, slot=1, note=n_a)
follower.on_block(b1)
assert follower.tip_state().leader_count == 1
# on fork, tip state is not updated
orphan = mk_block(genesis.block, slot=1, coin=c_b)
follower.on_block(orphan)
fork = mk_block(genesis.block, slot=1, note=n_b)
follower.on_block(fork)
assert follower.tip_state().block == b1
assert follower.tip_state().leader_count == 1
# after orphan is adopted, leader count should jumpy by 2 (each orphan counts as a leader)
b2 = mk_block(b1, slot=2, coin=c_a.evolve(), orphaned_proofs=[orphan])
# continuing the chain increments the leader count
b2 = mk_block(b1, slot=2, note=n_a)
follower.on_block(b2)
assert follower.tip_state().block == b2
assert follower.tip_state().leader_count == 3
assert follower.tip_state().leader_count == 2
def test_inference_on_empty_genesis_epoch(self):
coin = Coin(sk=0, value=10)
config = mk_config([coin]).replace(
note = Note(sk=0, value=10)
config = mk_config([note]).replace(
initial_total_active_stake=20,
total_active_stake_learning_rate=0.5,
active_slot_coeff=0.5,
)
genesis = mk_genesis_state([coin])
node = TestNode(config, genesis, coin)
genesis = mk_genesis_state([note])
node = TestNode(config, genesis, note)
# -- epoch 0 --
@ -77,12 +77,12 @@ class TestStakeRelativization(TestCase):
np.random.seed(seed)
stake = np.array((np.random.pareto(10, N) + 1) * 1000, dtype=np.int64)
coins = [Coin(sk=i, value=int(s)) for i, s in enumerate(stake)]
notes = [Note(sk=i, value=int(s)) for i, s in enumerate(stake)]
config = Config.cryptarchia_v0_0_1(stake.sum() * 2).replace(k=40)
genesis = mk_genesis_state(coins)
genesis = mk_genesis_state(notes)
nodes = [TestNode(config, genesis, c) for c in coins]
nodes = [TestNode(config, genesis, n) for n in notes]
T = config.epoch_length * EPOCHS
slot_leaders = np.zeros(T, dtype=np.int32)

219
cryptarchia/test_sync.py
View File

@ -1,22 +1,19 @@
from unittest import TestCase
from cryptarchia.cryptarchia import BlockHeader, Coin, Follower
from cryptarchia.cryptarchia import BlockHeader, Note, Follower
from cryptarchia.sync import InvalidBlockTree, sync
from cryptarchia.test_common import mk_block, mk_config, mk_genesis_state
from cryptarchia.test_common import mk_block, mk_chain, mk_config, mk_genesis_state
class TestSync(TestCase):
def test_sync_single_chain_from_genesis(self):
# Prepare a peer with a single chain:
# b0 - b1 - b2 - b3
coin = Coin(sk=0, value=10)
config = mk_config([coin])
genesis = mk_genesis_state([coin])
note = Note(sk=0, value=10)
config = mk_config([note])
genesis = mk_genesis_state([note])
peer = Follower(genesis, config)
b0, coin = mk_block(genesis.block, 1, coin), coin.evolve()
b1, coin = mk_block(b0, 2, coin), coin.evolve()
b2, coin = mk_block(b1, 3, coin), coin.evolve()
b3, coin = mk_block(b2, 4, coin), coin.evolve()
b0, b1, b2, b3 = mk_chain(genesis.block, note, slots=[1, 2, 3, 4])
for b in [b0, b1, b2, b3]:
peer.on_block(b)
self.assertEqual(peer.tip(), b3)
@ -32,14 +29,11 @@ class TestSync(TestCase):
def test_sync_single_chain_from_middle(self):
# Prepare a peer with a single chain:
# b0 - b1 - b2 - b3
coin = Coin(sk=0, value=10)
config = mk_config([coin])
genesis = mk_genesis_state([coin])
note = Note(sk=0, value=10)
config = mk_config([note])
genesis = mk_genesis_state([note])
peer = Follower(genesis, config)
b0, coin = mk_block(genesis.block, 1, coin), coin.evolve()
b1, coin = mk_block(b0, 2, coin), coin.evolve()
b2, coin = mk_block(b1, 3, coin), coin.evolve()
b3, coin = mk_block(b2, 4, coin), coin.evolve()
b0, b1, b2, b3 = mk_chain(genesis.block, note, slots=[1, 2, 3, 4])
for b in [b0, b1, b2, b3]:
peer.on_block(b)
self.assertEqual(peer.tip(), b3)
@ -61,18 +55,17 @@ class TestSync(TestCase):
# b0 - b1 - b2 - b5 == tip
# \
# b3 - b4
c_a, c_b = Coin(sk=0, value=10), Coin(sk=1, value=10)
config = mk_config([c_a, c_b])
genesis = mk_genesis_state([c_a, c_b])
n_a, n_b = Note(sk=0, value=10), Note(sk=1, value=10)
config = mk_config([n_a, n_b])
genesis = mk_genesis_state([n_a, n_b])
peer = Follower(genesis, config)
b0, c_a = mk_block(genesis.block, 1, c_a), c_a.evolve()
b1, c_a = mk_block(b0, 2, c_a), c_a.evolve()
b2, c_a = mk_block(b1, 3, c_a), c_a.evolve()
b3, c_b = mk_block(b0, 2, c_b), c_b.evolve()
b4, c_b = mk_block(b3, 3, c_b), c_b.evolve()
b5, c_a = mk_block(b2, 4, c_a), c_a.evolve()
b0, b1, b2, b5 = mk_chain(genesis.block, n_a, slots=[1, 2, 3, 4])
b3, b4 = mk_chain(b0, n_b, slots=[2, 3])
for b in [b0, b1, b2, b3, b4, b5]:
peer.on_block(b)
self.assertEqual(peer.tip(), b5)
self.assertEqual(peer.forks, [b4.id()])
@ -88,16 +81,14 @@ class TestSync(TestCase):
# b0 - b1 - b2 - b5 == tip
# \
# b3 - b4
c_a, c_b = Coin(sk=0, value=10), Coin(sk=1, value=10)
config = mk_config([c_a, c_b])
genesis = mk_genesis_state([c_a, c_b])
n_a, n_b = Note(sk=0, value=10), Note(sk=1, value=10)
config = mk_config([n_a, n_b])
genesis = mk_genesis_state([n_a, n_b])
peer = Follower(genesis, config)
b0, c_a = mk_block(genesis.block, 1, c_a), c_a.evolve()
b1, c_a = mk_block(b0, 2, c_a), c_a.evolve()
b2, c_a = mk_block(b1, 3, c_a), c_a.evolve()
b3, c_b = mk_block(b0, 2, c_b), c_b.evolve()
b4, c_b = mk_block(b3, 3, c_b), c_b.evolve()
b5, c_a = mk_block(b2, 4, c_a), c_a.evolve()
b0, b1, b2, b5 = mk_chain(genesis.block, n_a, slots=[1, 2, 3, 4])
b3, b4 = mk_chain(b0, n_b, slots=[2, 3])
for b in [b0, b1, b2, b3, b4, b5]:
peer.on_block(b)
self.assertEqual(peer.tip(), b5)
@ -121,16 +112,14 @@ class TestSync(TestCase):
# b0 - b1 - b2 - b5 == tip
# \
# b3 - b4
c_a, c_b = Coin(sk=0, value=10), Coin(sk=1, value=10)
config = mk_config([c_a, c_b])
genesis = mk_genesis_state([c_a, c_b])
n_a, n_b = Note(sk=0, value=10), Note(sk=1, value=10)
config = mk_config([n_a, n_b])
genesis = mk_genesis_state([n_a, n_b])
peer = Follower(genesis, config)
b0, c_a = mk_block(genesis.block, 1, c_a), c_a.evolve()
b1, c_a = mk_block(b0, 2, c_a), c_a.evolve()
b2, c_a = mk_block(b1, 3, c_a), c_a.evolve()
b3, c_b = mk_block(b0, 2, c_b), c_b.evolve()
b4, c_b = mk_block(b3, 3, c_b), c_b.evolve()
b5, c_a = mk_block(b2, 4, c_a), c_a.evolve()
b0, b1, b2, b5 = mk_chain(genesis.block, n_a, slots=[1, 2, 3, 4])
b3, b4 = mk_chain(b0, n_b, slots=[2, 3])
for b in [b0, b1, b2, b3, b4, b5]:
peer.on_block(b)
self.assertEqual(peer.tip(), b5)
@ -156,15 +145,13 @@ class TestSync(TestCase):
# Peer-1: b0 - b1 - b2
# \
# Peer-2: b3 - b4
c_a, c_b = Coin(sk=0, value=10), Coin(sk=1, value=10)
config = mk_config([c_a, c_b])
genesis = mk_genesis_state([c_a, c_b])
b0, c_a = mk_block(genesis.block, 1, c_a), c_a.evolve()
b1, c_a = mk_block(b0, 2, c_a), c_a.evolve()
b2, c_a = mk_block(b1, 3, c_a), c_a.evolve()
b3, c_b = mk_block(b0, 2, c_b), c_b.evolve()
b4, c_b = mk_block(b3, 3, c_b), c_b.evolve()
b5, c_a = mk_block(b2, 4, c_a), c_a.evolve()
n_a, n_b = Note(sk=0, value=10), Note(sk=1, value=10)
config = mk_config([n_a, n_b])
genesis = mk_genesis_state([n_a, n_b])
b0, b1, b2, b5 = mk_chain(genesis.block, n_a, slots=[1, 2, 3, 4])
b3, b4 = mk_chain(b0, n_b, slots=[2, 3])
peer0 = Follower(genesis, config)
for b in [b0, b1, b2, b5]:
peer0.on_block(b)
@ -200,23 +187,22 @@ class TestSync(TestCase):
# b0 - b1 - b2 - b3 - (invalid_b4) - (invalid_b5)
#
# First, build a valid chain (b0 ~ b3):
coin = Coin(sk=0, value=10)
config = mk_config([coin])
genesis = mk_genesis_state([coin])
note = Note(sk=0, value=10)
config = mk_config([note])
genesis = mk_genesis_state([note])
peer = Follower(genesis, config)
b0, coin = mk_block(genesis.block, 1, coin), coin.evolve()
b1, coin = mk_block(b0, 2, coin), coin.evolve()
b2, coin = mk_block(b1, 3, coin), coin.evolve()
b3, coin = mk_block(b2, 4, coin), coin.evolve()
b0, b1, b2, b3 = mk_chain(genesis.block, note, slots=[1, 2, 3, 4])
for b in [b0, b1, b2, b3]:
peer.on_block(b)
self.assertEqual(peer.tip(), b3)
self.assertEqual(peer.forks, [])
# And deliberately, add invalid blocks (b4 ~ b5):
fake_coin = Coin(sk=1, value=10)
b4, fake_coin = mk_block(b3, 5, fake_coin), fake_coin.evolve()
b5, fake_coin = mk_block(b4, 6, fake_coin), fake_coin.evolve()
fake_note = Note(sk=1, value=10)
b4 = mk_block(b3, 5, fake_note)
b5 = mk_block(b4, 6, fake_note)
apply_invalid_block_to_ledger_state(peer, b4)
apply_invalid_block_to_ledger_state(peer, b5)
# the tip shouldn't be changed.
@ -239,25 +225,23 @@ class TestSync(TestCase):
# b2 - (invalid_b6) - (invalid_b7)
#
# First, build a valid chain (b0 ~ b5):
c_a, c_b = Coin(sk=0, value=10), Coin(sk=1, value=10)
config = mk_config([c_a, c_b])
genesis = mk_genesis_state([c_a, c_b])
n_a, n_b = Note(sk=0, value=10), Note(sk=1, value=10)
config = mk_config([n_a, n_b])
genesis = mk_genesis_state([n_a, n_b])
peer = Follower(genesis, config)
b0, c_a = mk_block(genesis.block, 1, c_a), c_a.evolve()
b1, c_a = mk_block(b0, 2, c_a), c_a.evolve()
b2, c_b = mk_block(b0, 2, c_b), c_b.evolve()
b3, c_a = mk_block(b1, 3, c_a), c_a.evolve()
b4, c_a = mk_block(b3, 4, c_a), c_a.evolve()
b5, c_a = mk_block(b4, 5, c_a), c_a.evolve()
b0, b1, b3, b4, b5 = mk_chain(genesis.block, n_a, slots=[1, 2, 3, 4, 5])
b2 = mk_block(b0, 2, n_b)
for b in [b0, b1, b2, b3, b4, b5]:
peer.on_block(b)
self.assertEqual(peer.tip(), b5)
self.assertEqual(peer.forks, [b2.id()])
# And deliberately, add invalid blocks (b6 ~ b7):
fake_coin = Coin(sk=2, value=10)
b6, fake_coin = mk_block(b2, 3, fake_coin), fake_coin.evolve()
b7, fake_coin = mk_block(b6, 4, fake_coin), fake_coin.evolve()
fake_note = Note(sk=2, value=10)
b6, b7 = mk_chain(b2, fake_note, slots=[3, 4])
apply_invalid_block_to_ledger_state(peer, b6)
apply_invalid_block_to_ledger_state(peer, b7)
# the tip shouldn't be changed.
@ -287,14 +271,13 @@ class TestSyncFromCheckpoint(TestCase):
# b0 - b1 - b2 - b3
# ||
# checkpoint
coin = Coin(sk=0, value=10)
config = mk_config([coin])
genesis = mk_genesis_state([coin])
note = Note(sk=0, value=10)
config = mk_config([note])
genesis = mk_genesis_state([note])
peer = Follower(genesis, config)
b0, coin = mk_block(genesis.block, 1, coin), coin.evolve()
b1, coin = mk_block(b0, 2, coin), coin.evolve()
b2, coin = mk_block(b1, 3, coin), coin.evolve()
b3, coin = mk_block(b2, 4, coin), coin.evolve()
b0, b1, b2, b3 = mk_chain(genesis.block, note, slots=[1, 2, 3, 4])
for b in [b0, b1, b2, b3]:
peer.on_block(b)
self.assertEqual(peer.tip(), b3)
@ -324,16 +307,14 @@ class TestSyncFromCheckpoint(TestCase):
# b0 - b1 - b2 - b5 == tip
# \
# b3 - b4
c_a, c_b = Coin(sk=0, value=10), Coin(sk=1, value=10)
config = mk_config([c_a, c_b])
genesis = mk_genesis_state([c_a, c_b])
n_a, n_b = Note(sk=0, value=10), Note(sk=1, value=10)
config = mk_config([n_a, n_b])
genesis = mk_genesis_state([n_a, n_b])
peer = Follower(genesis, config)
b0, c_a = mk_block(genesis.block, 1, c_a), c_a.evolve()
b1, c_a = mk_block(b0, 2, c_a), c_a.evolve()
b2, c_a = mk_block(b1, 3, c_a), c_a.evolve()
b3, c_b = mk_block(b0, 2, c_b), c_b.evolve()
b4, c_b = mk_block(b3, 3, c_b), c_b.evolve()
b5, c_a = mk_block(b2, 4, c_a), c_a.evolve()
b0, b1, b2, b5 = mk_chain(genesis.block, n_a, slots=[1, 2, 3, 4])
b3, b4 = mk_chain(b0, n_b, slots=[2, 3])
for b in [b0, b1, b2, b3, b4, b5]:
peer.on_block(b)
self.assertEqual(peer.tip(), b5)
@ -363,15 +344,13 @@ class TestSyncFromCheckpoint(TestCase):
# Peer1: b3 - b4
# ||
# checkpoint
c_a, c_b = Coin(sk=0, value=10), Coin(sk=1, value=10)
config = mk_config([c_a, c_b])
genesis = mk_genesis_state([c_a, c_b])
b0, c_a = mk_block(genesis.block, 1, c_a), c_a.evolve()
b1, c_a = mk_block(b0, 2, c_a), c_a.evolve()
b2, c_a = mk_block(b1, 3, c_a), c_a.evolve()
b3, c_b = mk_block(b0, 2, c_b), c_b.evolve()
b4, c_b = mk_block(b3, 3, c_b), c_b.evolve()
b5, c_a = mk_block(b2, 4, c_a), c_a.evolve()
n_a, n_b = Note(sk=0, value=10), Note(sk=1, value=10)
config = mk_config([n_a, n_b])
genesis = mk_genesis_state([n_a, n_b])
b0, b1, b2, b5 = mk_chain(genesis.block, n_a, slots=[1, 2, 3, 4])
b3, b4 = mk_chain(b0, n_b, slots=[2, 3])
peer0 = Follower(genesis, config)
for b in [b0, b1, b2, b5]:
peer0.on_block(b)
@ -407,25 +386,22 @@ class TestSyncFromCheckpoint(TestCase):
# b2 - (invalid_b6) - (invalid_b7)
#
# First, build a valid chain (b0 ~ b5):
c_a, c_b = Coin(sk=0, value=10), Coin(sk=1, value=10)
config = mk_config([c_a, c_b])
genesis = mk_genesis_state([c_a, c_b])
n_a, n_b = Note(sk=0, value=10), Note(sk=1, value=10)
config = mk_config([n_a, n_b])
genesis = mk_genesis_state([n_a, n_b])
peer = Follower(genesis, config)
b0, c_a = mk_block(genesis.block, 1, c_a), c_a.evolve()
b1, c_a = mk_block(b0, 2, c_a), c_a.evolve()
b2, c_b = mk_block(b0, 2, c_b), c_b.evolve()
b3, c_a = mk_block(b1, 3, c_a), c_a.evolve()
b4, c_a = mk_block(b3, 4, c_a), c_a.evolve()
b5, c_a = mk_block(b4, 5, c_a), c_a.evolve()
b0, b1, b3, b4, b5 = mk_chain(genesis.block, n_a, slots=[1, 2, 3, 4, 5])
b2 = mk_block(b0, 2, n_b)
for b in [b0, b1, b2, b3, b4, b5]:
peer.on_block(b)
self.assertEqual(peer.tip(), b5)
self.assertEqual(peer.forks, [b2.id()])
# And deliberately, add invalid blocks (b6 ~ b7):
fake_coin = Coin(sk=2, value=10)
b6, fake_coin = mk_block(b2, 3, fake_coin), fake_coin.evolve()
b7, fake_coin = mk_block(b6, 4, fake_coin), fake_coin.evolve()
fake_note = Note(sk=2, value=10)
b6, b7 = mk_chain(b2, fake_note, slots=[3, 4])
apply_invalid_block_to_ledger_state(peer, b6)
apply_invalid_block_to_ledger_state(peer, b7)
# the tip shouldn't be changed.
@ -457,25 +433,22 @@ class TestSyncFromCheckpoint(TestCase):
# b2 - (invalid_b6) - (invalid_b7)
#
# First, build a valid chain (b0 ~ b5):
c_a, c_b = Coin(sk=0, value=10), Coin(sk=1, value=10)
config = mk_config([c_a, c_b])
genesis = mk_genesis_state([c_a, c_b])
n_a, n_b = Note(sk=0, value=10), Note(sk=1, value=10)
config = mk_config([n_a, n_b])
genesis = mk_genesis_state([n_a, n_b])
peer = Follower(genesis, config)
b0, c_a = mk_block(genesis.block, 1, c_a), c_a.evolve()
b1, c_a = mk_block(b0, 2, c_a), c_a.evolve()
b2, c_b = mk_block(b0, 2, c_b), c_b.evolve()
b3, c_a = mk_block(b1, 3, c_a), c_a.evolve()
b4, c_a = mk_block(b3, 4, c_a), c_a.evolve()
b5, c_a = mk_block(b4, 5, c_a), c_a.evolve()
b0, b1, b3, b4, b5 = mk_chain(genesis.block, n_a, slots=[1, 2, 3, 4, 5])
b2 = mk_block(b0, 2, n_b)
for b in [b0, b1, b2, b3, b4, b5]:
peer.on_block(b)
self.assertEqual(peer.tip(), b5)
self.assertEqual(peer.forks, [b2.id()])
# And deliberately, add invalid blocks (b6 ~ b7):
fake_coin = Coin(sk=2, value=10)
b6, fake_coin = mk_block(b2, 3, fake_coin), fake_coin.evolve()
b7, fake_coin = mk_block(b6, 4, fake_coin), fake_coin.evolve()
fake_note = Note(sk=2, value=10)
b6, b7 = mk_chain(b2, fake_note, slots=[3, 4])
apply_invalid_block_to_ledger_state(peer, b6)
apply_invalid_block_to_ledger_state(peer, b7)
# the tip shouldn't be changed.