[WIP] Attestation pool take 1: interop

chain/build.gradle

@@ -10,10 +10,20 @@ dependencies {
 
   implementation 'com.google.guava:guava'
   implementation 'io.projectreactor:reactor-core'
+  api "org.apache.commons:commons-collections4"
 
   testImplementation 'org.mockito:mockito-core'
+  testCompile 'io.projectreactor:reactor-test'
 
   // Gradle does not import test sources alongside with main sources
   // use a workaround until better solution will be found
   testImplementation project(':consensus').sourceSets.test.output
+
+  testImplementation 'org.junit.jupiter:junit-jupiter:5.5.1'
+
+  test {
+    useJUnitPlatform {
+      excludeTags 'FIX'
+    }
+  }
 }

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

@@ -1,5 +1,6 @@
 package org.ethereum.beacon.chain;
 
+import org.ethereum.beacon.core.state.Checkpoint;
 import org.reactivestreams.Publisher;
 
 public interface BeaconChain {
@@ -15,5 +16,22 @@ public interface BeaconChain {
    */
   BeaconTuple getRecentlyProcessed();
 
+  /**
+   * Returns the most recent justified checkpoint.
+   *
+   * @return a checkpoint.
+   */
+  Publisher<Checkpoint> getJustifiedCheckpoints();
+
+  /**
+   * Returns the most recent finalized checkpoint.
+   *
+   * <p><b>Note:</b> finalized checkpoints are published by {@link #getJustifiedCheckpoints()}
+   * either.
+   *
+   * @return a checkpoint.
+   */
+  Publisher<Checkpoint> getFinalizedCheckpoints();
+
   void init();
 }

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

@@ -8,15 +8,16 @@
 import org.apache.logging.log4j.Logger;
 import org.ethereum.beacon.chain.storage.BeaconChainStorage;
 import org.ethereum.beacon.chain.storage.BeaconTupleStorage;
+import org.ethereum.beacon.consensus.BeaconChainSpec;
 import org.ethereum.beacon.consensus.BeaconStateEx;
 import org.ethereum.beacon.consensus.BlockTransition;
-import org.ethereum.beacon.consensus.BeaconChainSpec;
 import org.ethereum.beacon.consensus.transition.EmptySlotTransition;
 import org.ethereum.beacon.consensus.verifier.BeaconBlockVerifier;
 import org.ethereum.beacon.consensus.verifier.BeaconStateVerifier;
 import org.ethereum.beacon.consensus.verifier.VerificationResult;
 import org.ethereum.beacon.core.BeaconBlock;
 import org.ethereum.beacon.core.BeaconState;
+import org.ethereum.beacon.core.state.Checkpoint;
 import org.ethereum.beacon.core.types.SlotNumber;
 import org.ethereum.beacon.schedulers.Schedulers;
 import org.ethereum.beacon.stream.SimpleProcessor;
@@ -36,6 +37,8 @@ public class DefaultBeaconChain implements MutableBeaconChain {
   private final BeaconTupleStorage tupleStorage;
 
   private final SimpleProcessor<BeaconTupleDetails> blockStream;
+  private final SimpleProcessor<Checkpoint> justifiedCheckpointStream;
+  private final SimpleProcessor<Checkpoint> finalizedCheckpointStream;
   private final Schedulers schedulers;
 
   private BeaconTuple recentlyProcessed;
@@ -58,6 +61,10 @@ public DefaultBeaconChain(
     this.schedulers = schedulers;
 
     blockStream = new SimpleProcessor<>(schedulers.events(), "DefaultBeaconChain.block");
+    justifiedCheckpointStream =
+        new SimpleProcessor<>(schedulers.events(), "DefaultBeaconChain.justifiedCheckpoint");
+    finalizedCheckpointStream =
+        new SimpleProcessor<>(schedulers.events(), "DefaultBeaconChain.finalizedCheckpoint");
   }
 
   @Override
@@ -66,6 +73,8 @@ public void init() {
       throw new IllegalStateException("Couldn't start from empty storage");
     }
     this.recentlyProcessed = fetchRecentTuple();
+    justifiedCheckpointStream.onNext(fetchJustifiedCheckpoint());
+    finalizedCheckpointStream.onNext(fetchFinalizedCheckpoint());
     blockStream.onNext(new BeaconTupleDetails(recentlyProcessed));
   }
 
@@ -117,7 +126,7 @@ public synchronized ImportResult insert(BeaconBlock block) {
 
     BeaconTuple newTuple = BeaconTuple.of(block, postBlockState);
     tupleStorage.put(newTuple);
-    updateFinality(parentState, postBlockState);
+    updateFinality(postBlockState);
 
     chainStorage.commit();
 
@@ -144,15 +153,49 @@ public BeaconTuple getRecentlyProcessed() {
     return recentlyProcessed;
   }
 
-  private void updateFinality(BeaconState previous, BeaconState current) {
-    if (!previous.getFinalizedCheckpoint().equals(current.getFinalizedCheckpoint())) {
+  private void updateFinality(BeaconState current) {
+    boolean finalizedStorageUpdated = false;
+    boolean justifiedStorageUpdated = false;
+    if (current
+        .getFinalizedCheckpoint()
+        .getEpoch()
+        .greater(fetchFinalizedCheckpoint().getEpoch())) {
       chainStorage.getFinalizedStorage().set(current.getFinalizedCheckpoint());
+      finalizedStorageUpdated = true;
     }
-    if (!previous.getCurrentJustifiedCheckpoint().equals(current.getCurrentJustifiedCheckpoint())) {
+    if (current
+        .getCurrentJustifiedCheckpoint()
+        .getEpoch()
+        .greater(fetchJustifiedCheckpoint().getEpoch())) {
       chainStorage.getJustifiedStorage().set(current.getCurrentJustifiedCheckpoint());
+      justifiedStorageUpdated = true;
+    }
+    // publish updates after both storages have been updated
+    // the order can be important if a finalizedCheckpointStream subscriber will look
+    // into justified storage
+    // in general, it may be important to publish after commit has succeeded
+    if (finalizedStorageUpdated) {
+      finalizedCheckpointStream.onNext(current.getFinalizedCheckpoint());
+    }
+    if (justifiedStorageUpdated) {
+      justifiedCheckpointStream.onNext(current.getCurrentJustifiedCheckpoint());
     }
   }
 
+  private Checkpoint fetchJustifiedCheckpoint() {
+    return chainStorage
+        .getJustifiedStorage()
+        .get()
+        .orElseThrow(() -> new RuntimeException("Justified checkpoint not found"));
+  }
+
+  private Checkpoint fetchFinalizedCheckpoint() {
+    return chainStorage
+        .getFinalizedStorage()
+        .get()
+        .orElseThrow(() -> new RuntimeException("Finalized checkpoint not found"));
+  }
+
   private BeaconStateEx pullParentState(BeaconBlock block) {
     Optional<BeaconTuple> parent = tupleStorage.get(block.getParentRoot());
     checkArgument(parent.isPresent(), "No parent for block %s", block);
@@ -187,4 +230,14 @@ private boolean rejectedByTime(BeaconBlock block) {
   public Publisher<BeaconTupleDetails> getBlockStatesStream() {
     return blockStream;
   }
+
+  @Override
+  public Publisher<Checkpoint> getJustifiedCheckpoints() {
+    return justifiedCheckpointStream;
+  }
+
+  @Override
+  public Publisher<Checkpoint> getFinalizedCheckpoints() {
+    return finalizedCheckpointStream;
+  }
 }

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

@@ -0,0 +1,7 @@
+package org.ethereum.beacon.chain;
+
+import org.reactivestreams.Publisher;
+
+public interface ForkChoiceProcessor {
+  Publisher<BeaconChainHead> getChainHeads();
+}

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

@@ -0,0 +1,183 @@
+package org.ethereum.beacon.chain;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import org.ethereum.beacon.chain.storage.BeaconChainStorage;
+import org.ethereum.beacon.consensus.BeaconChainSpec;
+import org.ethereum.beacon.consensus.spec.ForkChoice.LatestMessage;
+import org.ethereum.beacon.consensus.spec.ForkChoice.Store;
+import org.ethereum.beacon.core.BeaconBlock;
+import org.ethereum.beacon.core.BeaconState;
+import org.ethereum.beacon.core.operations.Attestation;
+import org.ethereum.beacon.core.operations.attestation.AttestationData;
+import org.ethereum.beacon.core.operations.slashing.IndexedAttestation;
+import org.ethereum.beacon.core.state.Checkpoint;
+import org.ethereum.beacon.core.types.SlotNumber;
+import org.ethereum.beacon.core.types.ValidatorIndex;
+import org.ethereum.beacon.schedulers.Schedulers;
+import org.ethereum.beacon.stream.SimpleProcessor;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.scheduler.Scheduler;
+import tech.pegasys.artemis.ethereum.core.Hash32;
+
+public class LMDGhostProcessor implements ForkChoiceProcessor {
+
+  private final int SEARCH_LIMIT = Integer.MAX_VALUE;
+
+  private final BeaconChainSpec spec;
+  private final BeaconChainStorage storage;
+  private final SimpleProcessor<BeaconChainHead> chainHeadStream;
+
+  private final Map<ValidatorIndex, LatestMessage> latestMessageStorage = new HashMap<>();
+  private Checkpoint justifiedCheckpoint = Checkpoint.EMPTY;
+  private Hash32 currentHeadRoot = Hash32.ZERO;
+
+  public LMDGhostProcessor(
+      BeaconChainSpec spec,
+      BeaconChainStorage storage,
+      Schedulers schedulers,
+      Publisher<Checkpoint> justifiedCheckpoints,
+      Publisher<IndexedAttestation> wireAttestations,
+      Publisher<? extends BeaconTuple> importedBlocks) {
+    this.spec = spec;
+    this.storage = storage;
+
+    Scheduler scheduler = schedulers.newSingleThreadDaemon("lmd-ghost-processor").toReactor();
+    this.chainHeadStream = new SimpleProcessor<>(scheduler, "LMDGhostProcessor.chainHeadStream");
+
+    Flux.from(justifiedCheckpoints).publishOn(scheduler).subscribe(this::onNewJustifiedCheckpoint);
+    Flux.from(wireAttestations).publishOn(scheduler).subscribe(this::onNewAttestation);
+    Flux.from(importedBlocks).publishOn(scheduler).subscribe(this::onNewImportedBlock);
+  }
+
+  private void onNewImportedBlock(BeaconTuple tuple) {
+    if (!isJustifiedAncestor(tuple.getBlock())) {
+      return;
+    }
+
+    for (Attestation attestation : tuple.getBlock().getBody().getAttestations()) {
+      List<ValidatorIndex> indices =
+          spec.get_attesting_indices(
+              tuple.getState(), attestation.getData(), attestation.getAggregationBits());
+      processAttestation(indices, attestation.getData());
+    }
+
+    updateHead();
+  }
+
+  private boolean isJustifiedAncestor(BeaconBlock block) {
+    // genesis shortcut
+    if (justifiedCheckpoint.equals(Checkpoint.EMPTY) && block.getSlot().equals(SlotNumber.ZERO)) {
+      return true;
+    }
+
+    BeaconBlock ancestor = block;
+    while (spec.compute_epoch_of_slot(ancestor.getSlot())
+        .greaterEqual(justifiedCheckpoint.getEpoch())) {
+      Optional<BeaconBlock> parent = storage.getBlockStorage().get(ancestor.getParentRoot());
+      if (!parent.isPresent()) {
+        return false;
+      }
+      if (parent.get().getParentRoot().equals(justifiedCheckpoint.getRoot())) {
+        return true;
+      }
+      ancestor = parent.get();
+    }
+
+    return false;
+  }
+
+  private void onNewAttestation(IndexedAttestation attestation) {
+    List<ValidatorIndex> indices = new ArrayList<>(attestation.getCustodyBit0Indices().listCopy());
+    indices.addAll(attestation.getCustodyBit1Indices().listCopy());
+    processAttestation(indices, attestation.getData());
+    updateHead();
+  }
+
+  private void processAttestation(List<ValidatorIndex> indices, AttestationData data) {
+    LatestMessage message =
+        new LatestMessage(data.getTarget().getEpoch(), data.getBeaconBlockRoot());
+    indices.forEach(
+        index -> {
+          latestMessageStorage.merge(
+              index,
+              message,
+              (oldMessage, newMessage) -> {
+                if (newMessage.getEpoch().greater(oldMessage.getEpoch())) {
+                  return newMessage;
+                } else {
+                  return oldMessage;
+                }
+              });
+        });
+  }
+
+  private void updateHead() {
+    Hash32 newHeadRoot = getHeadRoot();
+    if (!newHeadRoot.equals(currentHeadRoot)) {
+      BeaconTuple tuple = storage.getTupleStorage().get(newHeadRoot).get();
+      currentHeadRoot = newHeadRoot;
+      chainHeadStream.onNext(new BeaconChainHead(tuple));
+    }
+  }
+
+  private Hash32 getHeadRoot() {
+    return spec.get_head(
+        new Store() {
+
+          @Override
+          public Checkpoint getJustifiedCheckpoint() {
+            return storage.getJustifiedStorage().get().get();
+          }
+
+          @Override
+          public Checkpoint getFinalizedCheckpoint() {
+            return storage.getFinalizedStorage().get().get();
+          }
+
+          @Override
+          public Optional<BeaconBlock> getBlock(Hash32 root) {
+            return storage.getBlockStorage().get(root);
+          }
+
+          @Override
+          public Optional<BeaconState> getState(Hash32 root) {
+            return storage.getStateStorage().get(root);
+          }
+
+          @Override
+          public Optional<LatestMessage> getLatestMessage(ValidatorIndex index) {
+            return Optional.ofNullable(latestMessageStorage.get(index));
+          }
+
+          @Override
+          public List<Hash32> getChildren(Hash32 root) {
+            return storage.getBlockStorage().getChildren(root, SEARCH_LIMIT).stream()
+                .map(spec::signing_root)
+                .collect(Collectors.toList());
+          }
+        });
+  }
+
+  private void onNewJustifiedCheckpoint(Checkpoint checkpoint) {
+    if (checkpoint.getEpoch().greater(justifiedCheckpoint.getEpoch())) {
+      justifiedCheckpoint = checkpoint;
+      resetLatestMessages();
+      updateHead();
+    }
+  }
+
+  private void resetLatestMessages() {
+    latestMessageStorage.clear();
+  }
+
+  @Override
+  public Publisher<BeaconChainHead> getChainHeads() {
+    return chainHeadStream;
+  }
+}