Feature/wire

chain/src/main/java/org/ethereum/beacon/chain/BeaconTupleDetails.java

@@ -38,4 +38,8 @@ public BeaconStateEx getFinalState() {
     return getState();
   }
 
+  @Override
+  public String toString() {
+    return getFinalState().toString();
+  }
 }

chain/src/main/java/org/ethereum/beacon/chain/DefaultBeaconChain.java

@@ -98,17 +98,17 @@ private void initializeStorage() {
   }
 
   @Override
-  public synchronized boolean insert(BeaconBlock block) {
+  public synchronized ImportResult insert(BeaconBlock block) {
     if (rejectedByTime(block)) {
-      return false;
+      return ImportResult.ExpiredBlock;
     }
 
     if (exist(block)) {
-      return false;
+      return ImportResult.ExistingBlock;
     }
 
     if (!hasParent(block)) {
-      return false;
+      return ImportResult.NoParent;
     }
 
     long s = System.nanoTime();
@@ -121,7 +121,7 @@ public synchronized boolean insert(BeaconBlock block) {
     if (!blockVerification.isPassed()) {
       logger.warn("Block verification failed: " + blockVerification + ": " +
           block.toString(spec.getConstants(), parentState.getGenesisTime(), spec::signing_root));
-      return false;
+      return ImportResult.InvalidBlock;
     }
 
     BeaconStateEx postBlockState = blockTransition.apply(preBlockState, block);
@@ -130,7 +130,7 @@ public synchronized boolean insert(BeaconBlock block) {
         stateVerifier.verify(postBlockState, block);
     if (!stateVerification.isPassed()) {
       logger.warn("State verification failed: " + stateVerification);
-      return false;
+      return ImportResult.StateMismatch;
     }
 
     BeaconTuple newTuple = BeaconTuple.of(block, postBlockState);
@@ -154,7 +154,7 @@ public synchronized boolean insert(BeaconBlock block) {
                 spec::signing_root),
         String.format("%.3f", ((double) total) / 1_000_000_000d));
 
-    return true;
+    return ImportResult.OK;
   }
 
   @Override

chain/src/main/java/org/ethereum/beacon/chain/MutableBeaconChain.java

@@ -4,11 +4,21 @@
 
 public interface MutableBeaconChain extends BeaconChain {
 
+  enum ImportResult {
+    OK,
+    ExistingBlock,
+    NoParent,
+    ExpiredBlock,
+    InvalidBlock,
+    StateMismatch,
+    UnexpectedError
+  }
+
   /**
    * Inserts new block into a chain.
    *
    * @param block a block.
    * @return whether a block was inserted or not.
    */
-  boolean insert(BeaconBlock block);
+  ImportResult insert(BeaconBlock block);
 }

chain/src/main/java/org/ethereum/beacon/chain/ProposedBlockProcessorImpl.java

@@ -1,5 +1,6 @@
 package org.ethereum.beacon.chain;
 
+import org.ethereum.beacon.chain.MutableBeaconChain.ImportResult;
 import org.ethereum.beacon.core.BeaconBlock;
 import org.ethereum.beacon.schedulers.Schedulers;
 import org.ethereum.beacon.stream.SimpleProcessor;
@@ -19,10 +20,10 @@ public ProposedBlockProcessorImpl(MutableBeaconChain beaconChain, Schedulers sch
   }
 
   @Override
-  public void newBlockProposed(BeaconBlock newBlcok) {
-    boolean result = beaconChain.insert(newBlcok);
-    if (result) {
-      blocksStream.onNext(newBlcok);
+  public void newBlockProposed(BeaconBlock newBlock) {
+    ImportResult result = beaconChain.insert(newBlock);
+    if (result == ImportResult.OK) {
+      blocksStream.onNext(newBlock);
     }
   }
 

chain/src/main/java/org/ethereum/beacon/chain/observer/ObservableStateProcessorImpl.java

@@ -11,6 +11,8 @@
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.stream.Collectors;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.ethereum.beacon.chain.BeaconChainHead;
 import org.ethereum.beacon.chain.BeaconTuple;
 import org.ethereum.beacon.chain.BeaconTupleDetails;
@@ -39,7 +41,12 @@
 import reactor.core.publisher.Flux;
 
 public class ObservableStateProcessorImpl implements ObservableStateProcessor {
+  private static final Logger logger = LogManager.getLogger(ObservableStateProcessorImpl.class);
+
   private static final int MAX_TUPLE_CACHE_SIZE = 32;
+
+  private final int maxEmptySlotTransitions = 256;
+
   private final BeaconTupleStorage tupleStorage;
 
   private final HeadFunction headFunction;
@@ -218,22 +225,40 @@ private void newHead(BeaconTupleDetails head) {
 
   private void newSlot(SlotNumber newSlot) {
     if (head.getBlock().getSlot().greater(newSlot)) {
+      logger.info("Ignore new slot " + newSlot + " below head block: " + head.getBlock());
+      return;
+    }
+    if (newSlot.greater(head.getBlock().getSlot().plus(maxEmptySlotTransitions))) {
+      logger.debug("Ignore new slot " + newSlot + " far above head block: " + head.getBlock());
       return;
     }
+
     updateCurrentObservableState(head, newSlot);
   }
 
   private void updateCurrentObservableState(BeaconTupleDetails head, SlotNumber slot) {
     assert slot.greaterEqual(head.getBlock().getSlot());
 
-    PendingOperations pendingOperations =
-        getPendingOperations(head.getFinalState(), copyAttestationCache());
     if (slot.greater(head.getBlock().getSlot())) {
-      BeaconStateEx stateUponASlot = emptySlotTransition.apply(head.getFinalState(), slot);
+      BeaconStateEx stateUponASlot;
+      if (latestState.getSlot().greater(spec.getConstants().getGenesisSlot())
+          && spec.getObjectHasher()
+              .getHashTruncateLast(head.getBlock())
+              .equals(
+                  spec.get_block_root_at_slot(latestState, latestState.getSlot().decrement()))) {
+
+        // latestState is actual with respect to current head
+        stateUponASlot = emptySlotTransition.apply(latestState, slot);
+      } else {
+        // recalculate all empty slots starting from the head
+        stateUponASlot = emptySlotTransition.apply(head.getFinalState(), slot);
+      }
       latestState = stateUponASlot;
+      PendingOperations pendingOperations = getPendingOperations(stateUponASlot, copyAttestationCache());
       observableStateStream.onNext(
           new ObservableBeaconState(head.getBlock(), stateUponASlot, pendingOperations));
     } else {
+      PendingOperations pendingOperations = getPendingOperations(head.getFinalState(), copyAttestationCache());
       if (head.getPostSlotState().isPresent()) {
         latestState = head.getPostSlotState().get();
         observableStateStream.onNext(new ObservableBeaconState(
@@ -255,16 +280,9 @@ private PendingOperations getPendingOperations(
       BeaconState state, Map<BLSPubkey, List<Attestation>> attestationMap) {
     List<Attestation> attestations = attestationMap.values().stream()
         .flatMap(Collection::stream)
-        .filter(attestation -> {
-          /* attestation_slot = get_attestation_slot(state, attestation)
-             assert attestation_slot + MIN_ATTESTATION_INCLUSION_DELAY <= state.slot <= attestation_slot + SLOTS_PER_EPOCH */
-          SlotNumber attestationSlot = spec.get_attestation_slot(state, attestation.getData());
-          SlotNumber lowerBoundary =
-              attestationSlot.plus(spec.getConstants().getMinAttestationInclusionDelay());
-          SlotNumber upperBoundary = attestationSlot.plus(spec.getConstants().getSlotsPerEpoch());
-          return lowerBoundary.lessEqual(state.getSlot())
-              && state.getSlot().lessEqual(upperBoundary);
-        })
+        .filter(attestation ->
+            attestation.getData().getTargetEpoch().lessEqual(spec.get_current_epoch(state)))
+        .filter(attestation -> spec.verify_attestation(state, attestation))
         .sorted(Comparator.comparing(attestation -> attestation.getData().getTargetEpoch()))
         .collect(Collectors.toList());
 

chain/src/main/java/org/ethereum/beacon/chain/observer/PendingOperationsState.java

@@ -53,7 +53,7 @@ public List<Attestation> peekAggregateAttestations(int maxCount) {
 
   private Attestation aggregateAttestations(List<Attestation> attestations) {
     assert !attestations.isEmpty();
-    assert attestations.stream().skip(1).allMatch(a -> a.equals(attestations.get(0)));
+    assert attestations.stream().skip(1).allMatch(a -> a.getData().equals(attestations.get(0).getData()));
 
     Bitfield participants =
         attestations.stream()

chain/src/main/java/org/ethereum/beacon/chain/storage/BeaconChainStorage.java

@@ -1,12 +1,16 @@
 package org.ethereum.beacon.chain.storage;
 
+import org.ethereum.beacon.core.BeaconBlockHeader;
+import org.ethereum.beacon.db.source.DataSource;
 import org.ethereum.beacon.db.source.SingleValueSource;
 import tech.pegasys.artemis.ethereum.core.Hash32;
 
 public interface BeaconChainStorage {
 
   BeaconBlockStorage getBlockStorage();
 
+  DataSource<Hash32, BeaconBlockHeader> getBlockHeaderStorage();
+
   BeaconStateStorage getStateStorage();
 
   BeaconTupleStorage getTupleStorage();

chain/src/main/java/org/ethereum/beacon/chain/storage/impl/BeaconChainStorageImpl.java

@@ -4,25 +4,30 @@
 import org.ethereum.beacon.chain.storage.BeaconChainStorage;
 import org.ethereum.beacon.chain.storage.BeaconStateStorage;
 import org.ethereum.beacon.chain.storage.BeaconTupleStorage;
+import org.ethereum.beacon.core.BeaconBlockHeader;
+import org.ethereum.beacon.db.source.DataSource;
 import org.ethereum.beacon.db.source.SingleValueSource;
 import tech.pegasys.artemis.ethereum.core.Hash32;
 
 /** A default implementation of {@link BeaconChainStorage}. */
 public class BeaconChainStorageImpl implements BeaconChainStorage {
 
   private final BeaconBlockStorage blockStorage;
+  private final DataSource<Hash32, BeaconBlockHeader> blockHeaderStorage;
   private final BeaconStateStorage stateStorage;
   private final BeaconTupleStorage tupleStorage;
   private final SingleValueSource<Hash32> justifiedStorage;
   private final SingleValueSource<Hash32> finalizedStorage;
 
   public BeaconChainStorageImpl(
       BeaconBlockStorage blockStorage,
+      DataSource<Hash32, BeaconBlockHeader> blockHeaderStorage,
       BeaconStateStorage stateStorage,
       BeaconTupleStorage tupleStorage,
       SingleValueSource<Hash32> justifiedStorage,
       SingleValueSource<Hash32> finalizedStorage) {
     this.blockStorage = blockStorage;
+    this.blockHeaderStorage = blockHeaderStorage;
     this.stateStorage = stateStorage;
     this.tupleStorage = tupleStorage;
     this.justifiedStorage = justifiedStorage;
@@ -34,6 +39,11 @@ public BeaconBlockStorage getBlockStorage() {
     return blockStorage;
   }
 
+  @Override
+  public DataSource<Hash32, BeaconBlockHeader> getBlockHeaderStorage() {
+    return blockHeaderStorage;
+  }
+
   @Override
   public BeaconStateStorage getStateStorage() {
     return stateStorage;

chain/src/main/java/org/ethereum/beacon/chain/storage/impl/DelegateBlockHeaderStorageImpl.java

@@ -0,0 +1,54 @@
+package org.ethereum.beacon.chain.storage.impl;
+
+import java.util.Optional;
+import javax.annotation.Nonnull;
+import org.ethereum.beacon.chain.storage.BeaconBlockStorage;
+import org.ethereum.beacon.consensus.hasher.ObjectHasher;
+import org.ethereum.beacon.core.BeaconBlock;
+import org.ethereum.beacon.core.BeaconBlockHeader;
+import org.ethereum.beacon.db.source.DataSource;
+import tech.pegasys.artemis.ethereum.core.Hash32;
+
+public class DelegateBlockHeaderStorageImpl implements DataSource<Hash32, BeaconBlockHeader> {
+
+  private final BeaconBlockStorage delegateBlockStorage;
+  private final ObjectHasher<Hash32> objectHasher;
+
+  public DelegateBlockHeaderStorageImpl(
+      BeaconBlockStorage delegateBlockStorage,
+      ObjectHasher<Hash32> objectHasher) {
+    this.delegateBlockStorage = delegateBlockStorage;
+    this.objectHasher = objectHasher;
+  }
+
+  @Override
+  public Optional<BeaconBlockHeader> get(@Nonnull Hash32 key) {
+    return delegateBlockStorage
+        .get(key)
+        .map(this::createHeader);
+  }
+
+  private BeaconBlockHeader createHeader(BeaconBlock block) {
+    return new BeaconBlockHeader(
+        block.getSlot(),
+        block.getPreviousBlockRoot(),
+        block.getStateRoot(),
+        objectHasher.getHash(block.getBody()),
+        block.getSignature());
+  }
+
+  @Override
+  public void put(@Nonnull Hash32 key, @Nonnull BeaconBlockHeader value) {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public void remove(@Nonnull Hash32 key) {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public void flush() {
+    throw new UnsupportedOperationException();
+  }
+}

chain/src/main/java/org/ethereum/beacon/chain/storage/impl/MemBeaconChainStorageFactory.java

@@ -27,7 +27,12 @@ public BeaconChainStorage create(Database database) {
         new BeaconStateStorageImpl(new HashMapDataSource<>(), objectHasher);
     BeaconTupleStorage tupleStorage = new BeaconTupleStorageImpl(blockStorage, stateStorage);
 
-    return new BeaconChainStorageImpl(blockStorage, stateStorage, tupleStorage,
-        SingleValueSource.memSource(), SingleValueSource.memSource());
+    return new BeaconChainStorageImpl(
+        blockStorage,
+        new DelegateBlockHeaderStorageImpl(blockStorage, objectHasher),
+        stateStorage,
+        tupleStorage,
+        SingleValueSource.memSource(),
+        SingleValueSource.memSource());
   }
 }

chain/src/main/java/org/ethereum/beacon/chain/storage/impl/SSZBeaconChainStorageFactory.java

@@ -41,7 +41,12 @@ public BeaconChainStorage create(Database database) {
     SingleValueSource<Hash32> finalizedStorage = createHash32Storage(database, "finalized-hash");
 
     return new BeaconChainStorageImpl(
-        blockStorage, stateStorage, tupleStorage, justifiedStorage, finalizedStorage);
+        blockStorage,
+        new DelegateBlockHeaderStorageImpl(blockStorage, objectHasher),
+        stateStorage,
+        tupleStorage,
+        justifiedStorage,
+        finalizedStorage);
   }
 
   private SingleValueSource<Hash32> createHash32Storage(Database database, String name) {

chain/src/main/java/org/ethereum/beacon/chain/storage/impl/SerializerFactory.java

@@ -13,8 +13,7 @@ public interface SerializerFactory {
   <T> Function<T, BytesValue> getSerializer(Class<T> objectClass);
 
   static SerializerFactory createSSZ(SpecConstants specConstants) {
-    return new SSZSerializerFactory(
-        new SSZBuilder()
+    return new SSZSerializerFactory(new SSZBuilder()
             .withExternalVarResolver(new SpecConstantsResolver(specConstants))
             .buildSerializer());
   }

chain/src/test/java/org/ethereum/beacon/chain/DefaultBeaconChainTest.java

@@ -2,6 +2,7 @@
 
 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;
 import org.ethereum.beacon.chain.storage.impl.SerializerFactory;
@@ -54,7 +55,7 @@ public void insertAChain() {
               BeaconTuple recentlyProcessed = beaconChain.getRecentlyProcessed();
               BeaconBlock aBlock = createBlock(recentlyProcessed, spec,
                   schedulers.getCurrentTime(), perSlotTransition);
-              Assert.assertTrue(beaconChain.insert(aBlock));
+              Assert.assertEquals(ImportResult.OK, beaconChain.insert(aBlock));
               Assert.assertEquals(aBlock, beaconChain.getRecentlyProcessed().getBlock());
 
               System.out.println("Inserted block: " + (idx + 1));