Skip to content

Commit

Permalink
Merge pull request #212 from harmony-dev/fix/fork-choice
Browse files Browse the repository at this point in the history 
Fix/fork choice
  • Loading branch information
@mkalinin
mkalinin authored Nov 30, 2019
2 parents bea450e + a97f025 commit 212f836
Show file tree
Hide file tree
Showing 5 changed files with 121 additions and 42 deletions.
14 changes: 10 additions & 4 deletions chain/src/main/java/org/ethereum/beacon/chain/DefaultBeaconChain.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -115,6 +115,13 @@ public synchronized ImportResult insert(BeaconBlock block) {

long s = System.nanoTime();

Hash32 finalizedRoot = chainStorage.getFinalizedStorage().get().get().getRoot();
SlotNumber finalizedSlot = chainStorage.getBlockStorage().get(finalizedRoot).get().getSlot();
Hash32 finalizedAncestor = getAncestor(spec.signing_root(block), block, finalizedSlot);
if (!finalizedAncestor.equals(finalizedRoot)) {
return ImportResult.ExpiredBlock;
}

BeaconStateEx parentState = pullParentState(block);

BeaconStateEx preBlockState = preBlockTransition.apply(parentState, block.getSlot());
Expand Down Expand Up @@ -240,10 +247,9 @@ private boolean hasParent(BeaconBlock block) {
* @return true if block should be rejected, false otherwise.
*/
private boolean rejectedByTime(BeaconBlock block) {
SlotNumber nextToCurrentSlot =
spec.get_current_slot(recentlyProcessed.getState(), schedulers.getCurrentTime()).increment();

return block.getSlot().greater(nextToCurrentSlot);
SlotNumber currentSlot =
spec.get_current_slot(recentlyProcessed.getState(), schedulers.getCurrentTime());
return block.getSlot().greater(currentSlot);
}

@Override
Expand Down
87 changes: 60 additions & 27 deletions chain/src/main/java/org/ethereum/beacon/chain/observer/ObservableStateProcessorImpl.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -27,8 +27,11 @@
import org.ethereum.beacon.consensus.transition.EmptySlotTransition;
import org.ethereum.beacon.core.BeaconBlock;
import org.ethereum.beacon.core.BeaconState;
import org.ethereum.beacon.core.MutableBeaconState;
import org.ethereum.beacon.core.operations.Attestation;
import org.ethereum.beacon.core.state.Checkpoint;
import org.ethereum.beacon.core.operations.attestation.AttestationData;
import org.ethereum.beacon.core.operations.slashing.IndexedAttestation;
import org.ethereum.beacon.core.state.PendingAttestation;
import org.ethereum.beacon.core.types.EpochNumber;
import org.ethereum.beacon.core.types.SlotNumber;
Expand Down Expand Up @@ -67,7 +70,9 @@ public class ObservableStateProcessorImpl implements ObservableStateProcessor {
private Cache<BeaconBlock, BeaconTupleDetails> tupleDetails = new LRUCache<>(MAX_TUPLE_CACHE_SIZE);

private final List<Attestation> attestationBuffer = new ArrayList<>();
private final Map<Pair<ValidatorIndex, EpochNumber>, Attestation> attestationCache = new HashMap<>();

private final Map<Pair<ValidatorIndex, EpochNumber>, Attestation> offChainAttestations = new HashMap<>();
private final Map<ValidatorIndex, LatestMessage> latestMessages = new HashMap<>();
private final Schedulers schedulers;

private final SimpleProcessor<BeaconChainHead> headStream;
Expand Down Expand Up @@ -152,17 +157,46 @@ private void doHardWork() {
}
List<Attestation> attestations = drainAttestations(spec.get_current_epoch(latestState));
for (Attestation attestation : attestations) {

List<ValidatorIndex> participants =
spec.get_attesting_indices(
latestState, attestation.getData(), attestation.getAggregationBits());

participants.forEach(index -> addValidatorAttestation(index, attestation));
try {
BeaconTuple tuple = tupleStorage.get(attestation.getData().getTarget().getRoot()).get();

MutableBeaconState mutableState = tuple.getState().createMutableCopy();
spec.process_slots(
mutableState,
spec.compute_start_slot_at_epoch(attestation.getData().getTarget().getEpoch()));
BeaconState refState = mutableState.createImmutable();
IndexedAttestation indexed_attestation =
spec.get_indexed_attestation(refState, attestation);
if (!spec.is_valid_indexed_attestation(refState, indexed_attestation)) {
continue;
}

List<ValidatorIndex> participants =
spec.get_attesting_indices(
refState, attestation.getData(), attestation.getAggregationBits());

participants.forEach(index -> addValidatorAttestation(index, attestation));
} catch (RuntimeException e) {
continue;
}
}
if (attestations.size() > 0) {
updateHead(latestState);
}
}

private synchronized void addValidatorAttestation(ValidatorIndex index, Attestation attestation) {
attestationCache.put(Pair.with(index, attestation.getData().getTarget().getEpoch()), attestation);
updateLatestMessages(index, attestation.getData());
offChainAttestations.put(
Pair.with(index, attestation.getData().getTarget().getEpoch()), attestation);
}

private void updateLatestMessages(ValidatorIndex index, AttestationData data) {
EpochNumber targetEpoch = data.getTarget().getEpoch();
if (!latestMessages.containsKey(index)
|| targetEpoch.greater(latestMessages.get(index).getEpoch())) {
latestMessages.put(index, new LatestMessage(targetEpoch, data.getBeaconBlockRoot()));
}
}

private synchronized void onNewAttestation(Attestation attestation) {
Expand Down Expand Up @@ -205,29 +239,34 @@ private void addAttestationsFromState(BeaconState beaconState) {
EpochNumber targetEpoch = pendingAttestation.getData().getTarget().getEpoch();
participants.forEach(
index -> {
updateLatestMessages(index, pendingAttestation.getData());
removeValidatorAttestation(index, targetEpoch);
});
}
}

private synchronized void removeValidatorAttestation(ValidatorIndex index, EpochNumber epoch) {
attestationCache.remove(Pair.with(index, epoch));
offChainAttestations.remove(Pair.with(index, epoch));
}

/** Purges all entries for epochs before {@code targetEpoch}*/
private synchronized void purgeAttestations(EpochNumber targetEpoch) {
attestationCache.entrySet()
offChainAttestations.entrySet()
.removeIf(entry -> entry.getValue().getData().getTarget().getEpoch().less(targetEpoch));
}

private synchronized Map<ValidatorIndex, List<Attestation>> copyAttestationCache() {
return attestationCache.entrySet().stream()
private synchronized Map<ValidatorIndex, List<Attestation>> copyOffChainAttestations() {
return offChainAttestations.entrySet().stream()
.collect(
Collectors.groupingBy(
e -> e.getKey().getValue0(),
Collectors.mapping(Entry::getValue, Collectors.toList())));
}

private synchronized Map<ValidatorIndex, LatestMessage> copyLatestMessages() {
return new HashMap<>(latestMessages);
}

private BeaconTupleDetails head;
private BeaconStateEx latestState;

Expand Down Expand Up @@ -271,7 +310,10 @@ private void newSlot(SlotNumber newSlot) {
}
}

updateCurrentObservableState(head, newSlot);
updateHead(latestState);
if (newSlot.greater(latestState.getSlot())) {
updateCurrentObservableState(head, newSlot);
}
}

private void updateCurrentObservableState(BeaconTupleDetails head, SlotNumber slot) {
Expand All @@ -292,11 +334,11 @@ private void updateCurrentObservableState(BeaconTupleDetails head, SlotNumber sl
stateUponASlot = emptySlotTransition.apply(head.getFinalState(), slot);
}
latestState = stateUponASlot;
PendingOperations pendingOperations = getPendingOperations(stateUponASlot, copyAttestationCache());
PendingOperations pendingOperations = getPendingOperations(stateUponASlot, copyOffChainAttestations());
observableStateStream.onNext(
new ObservableBeaconState(head.getBlock(), stateUponASlot, pendingOperations));
} else {
PendingOperations pendingOperations = getPendingOperations(head.getFinalState(), copyAttestationCache());
PendingOperations pendingOperations = getPendingOperations(head.getFinalState(), copyOffChainAttestations());
if (head.getPostSlotState().isPresent()) {
latestState = head.getPostSlotState().get();
observableStateStream.onNext(new ObservableBeaconState(
Expand Down Expand Up @@ -328,20 +370,11 @@ private PendingOperations getPendingOperations(
}

private void updateHead(BeaconState state) {
Map<ValidatorIndex, List<Attestation>> attestationCacheCopy = copyAttestationCache();
Map<ValidatorIndex, LatestMessage> latestMessagesCopy = copyLatestMessages();
BeaconBlock newHead =
headFunction.getHead(
validatorIndex -> {
List<Attestation> validatorAttestations =
attestationCacheCopy.getOrDefault(validatorIndex, Collections.emptyList());

return validatorAttestations.stream()
.max(Comparator.comparing(attestation -> attestation.getData().getTarget().getEpoch()))
.flatMap(a -> Optional.of(
new LatestMessage(
spec.compute_epoch_at_slot(a.getData().getSlot()),
a.getData().getBeaconBlockRoot())));
});
validatorIndex ->
Optional.ofNullable(latestMessagesCopy.getOrDefault(validatorIndex, null)));
if (this.head != null && this.head.getBlock().equals(newHead)) {
return; // == old
}
Expand Down
55 changes: 50 additions & 5 deletions chain/src/test/java/org/ethereum/beacon/chain/DefaultBeaconChainTest.java
View file
Original file line number Diff line number Diff line change
@@ -1,7 +1,5 @@
package org.ethereum.beacon.chain;

import java.util.Collections;
import java.util.stream.IntStream;
import org.ethereum.beacon.chain.MutableBeaconChain.ImportResult;
import org.ethereum.beacon.chain.storage.BeaconChainStorage;
import org.ethereum.beacon.chain.storage.impl.SSZBeaconChainStorageFactory;
Expand All @@ -26,19 +24,26 @@
import org.ethereum.beacon.core.BeaconState;
import org.ethereum.beacon.core.state.Eth1Data;
import org.ethereum.beacon.core.types.BLSSignature;
import org.ethereum.beacon.core.types.SlotNumber;
import org.ethereum.beacon.core.types.Time;
import org.ethereum.beacon.db.Database;
import org.ethereum.beacon.schedulers.ControlledSchedulers;
import org.ethereum.beacon.schedulers.Schedulers;
import org.junit.Assert;
import org.junit.Test;
import tech.pegasys.artemis.ethereum.core.Hash32;
import tech.pegasys.artemis.util.uint.UInt64;

import java.time.Instant;
import java.time.temporal.ChronoUnit;
import java.util.Collections;
import java.util.stream.IntStream;

public class DefaultBeaconChainTest {

@Test
public void insertAChain() {
Schedulers schedulers = Schedulers.createDefault();
ControlledSchedulers schedulers = Schedulers.createControlled();

BeaconChainSpec spec =
BeaconChainSpec.Builder.createWithDefaultParams()
Expand All @@ -58,8 +63,12 @@ public void insertAChain() {
.forEach(
(idx) -> {
BeaconTuple recentlyProcessed = beaconChain.getRecentlyProcessed();
BeaconBlock aBlock = createBlock(recentlyProcessed, spec,
schedulers.getCurrentTime(), perSlotTransition);
schedulers.setCurrentTime(
spec.get_slot_start_time(recentlyProcessed.getState(),
recentlyProcessed.getBlock().getSlot().increment()).getValue()*1000);
BeaconBlock aBlock =
createBlock(
recentlyProcessed, spec, schedulers.getCurrentTime(), perSlotTransition);
Assert.assertEquals(ImportResult.OK, beaconChain.insert(aBlock));
Assert.assertEquals(aBlock, beaconChain.getRecentlyProcessed().getBlock());

Expand Down Expand Up @@ -118,4 +127,40 @@ public BeaconStateEx apply(BeaconStateEx stateEx) {
chainStorage,
schedulers);
}

@Test
public void testRejectBlocks() {
ControlledSchedulers schedulers = Schedulers.createControlled();
Instant genesisTime = Instant.now().plus(1, ChronoUnit.DAYS);
schedulers.setCurrentTime(genesisTime.toEpochMilli());

BeaconChainSpec spec =
BeaconChainSpec.Builder.createWithDefaultParams()
.withComputableGenesisTime(false)
.withVerifyDepositProof(false)
.build();
StateTransition<BeaconStateEx> perSlotTransition =
StateTransitionTestUtil.createNextSlotTransition();
MutableBeaconChain beaconChain = createBeaconChain(spec, perSlotTransition, schedulers);

beaconChain.init();
BeaconTuple initialTuple = beaconChain.getRecentlyProcessed();
Assert.assertEquals(spec.getConstants().getGenesisSlot(), initialTuple.getBlock().getSlot());

BeaconTuple recentlyProcessed = beaconChain.getRecentlyProcessed();

schedulers.setCurrentTime(
spec.get_slot_start_time(initialTuple.getState(), SlotNumber.of(2)).getValue() * 1000);

SlotNumber nextSlot =
spec.get_current_slot(recentlyProcessed.getState(), schedulers.getCurrentTime())
.plus(1);
long nextSlotTime =
spec.get_slot_start_time(recentlyProcessed.getState(), nextSlot).getMillis().getValue() + 1;

BeaconBlock aBlock =
createBlock(recentlyProcessed, spec, nextSlotTime, perSlotTransition);

Assert.assertEquals(ImportResult.ExpiredBlock, beaconChain.insert(aBlock));
}
}
1 change: 0 additions & 1 deletion core/src/main/java/org/ethereum/beacon/core/operations/Attestation.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -83,7 +83,6 @@ public Attestation withData(AttestationData data) {
}

public Attestation withSignature(BLSSignature signature) {
assert BLSSignature.ZERO.equals(signature);
return new Attestation(aggregationBits, data, signature);
}

Expand Down
6 changes: 1 addition & 5 deletions ...ore/src/main/java/org/ethereum/beacon/validator/attester/BeaconAttestationSignerImpl.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -32,10 +32,6 @@ public Attestation sign(Attestation attestation, BeaconState state) {
UInt64 domain = spec.get_domain(state, BEACON_ATTESTER, attestation.getData().getTarget().getEpoch());
BLSSignature signature = signer.sign(hash, domain);

return new Attestation(
attestation.getAggregationBits(),
attestation.getData(),
signature,
spec.getConstants());
return attestation.withSignature(signature);
}
}

0 comments on commit 212f836

Please sign in to comment.