[WIP] Attestation pool take 1: interop

chain/build.gradle

@@ -10,6 +10,7 @@ dependencies {
 
   implementation 'com.google.guava:guava'
   implementation 'io.projectreactor:reactor-core'
+  api "org.apache.commons:commons-collections4"
 
   testImplementation 'org.mockito:mockito-core'
 

chain/src/main/java/org/ethereum/beacon/chain/pool/AttestationPool.java

@@ -0,0 +1,135 @@
+package org.ethereum.beacon.chain.pool;
+
+import java.time.Duration;
+import org.ethereum.beacon.chain.BeaconChain;
+import org.ethereum.beacon.chain.pool.checker.SanityChecker;
+import org.ethereum.beacon.chain.pool.checker.SignatureEncodingChecker;
+import org.ethereum.beacon.chain.pool.checker.TimeFrameFilter;
+import org.ethereum.beacon.chain.pool.churn.OffChainAggregates;
+import org.ethereum.beacon.chain.pool.registry.ProcessedAttestations;
+import org.ethereum.beacon.chain.pool.registry.UnknownAttestationPool;
+import org.ethereum.beacon.chain.pool.verifier.AttestationVerifier;
+import org.ethereum.beacon.chain.pool.verifier.BatchVerifier;
+import org.ethereum.beacon.chain.storage.BeaconChainStorage;
+import org.ethereum.beacon.consensus.BeaconChainSpec;
+import org.ethereum.beacon.consensus.transition.EmptySlotTransition;
+import org.ethereum.beacon.core.BeaconBlock;
+import org.ethereum.beacon.core.state.Checkpoint;
+import org.ethereum.beacon.core.types.EpochNumber;
+import org.ethereum.beacon.core.types.SlotNumber;
+import org.ethereum.beacon.schedulers.Schedulers;
+import org.reactivestreams.Publisher;
+
+/**
+ * Attestation pool API.
+ *
+ * <p>Along with {@link BeaconChain} attestation pool is one of the central components of the Beacon
+ * chain client. Its main responsibilities are to verify attestation coming from the wire and
+ * accumulate those of them which are not yet included on chain.
+ *
+ * <p>A list of attestation pool clients:
+ *
+ * <ul>
+ *   <li>Wire - valid attestations should be further propagated; peers sent attestations that are
+ *       eventually invalid must be dropped.
+ *   <li>Fork choice - LMD GHOST is driven by attestations received from the wire even if they are
+ *       not yet on chain.
+ *   <li>Validator - attestations not yet included on chain should be included into a new block
+ *       produced by proposer.
+ * </ul>
+ */
+public interface AttestationPool {
+
+  /** A number of threads in the executor processing attestation pool. */
+  int MAX_THREADS = 32;
+
+  /** Discard attestations with target epoch greater than current epoch plus this number. */
+  EpochNumber MAX_ATTESTATION_LOOKAHEAD = EpochNumber.of(1);
+
+  /**
+   * Max number of attestations kept by processed attestations registry. An entry of this registry
+   * should be represented by a hash of registered attestation.
+   */
+  int MAX_PROCESSED_ATTESTATIONS = 1_000_000;
+
+  /** Max number of attestations made to not yet known block that could be kept in memory. */
+  int MAX_UNKNOWN_ATTESTATIONS = 100_000;
+
+  /** Max size of a buffer that collects attestations before passing them on the main verifier. */
+  int VERIFIER_BUFFER_SIZE = 10_000;
+
+  /** A throttling interval for verifier buffer. */
+  Duration VERIFIER_INTERVAL = Duration.ofMillis(50);
+
+  /**
+   * Valid attestations publisher.
+   *
+   * @return a publisher.
+   */
+  Publisher<ReceivedAttestation> getValid();
+
+  /**
+   * Invalid attestations publisher.
+   *
+   * @return a publisher.
+   */
+  Publisher<ReceivedAttestation> getInvalid();
+
+  /**
+   * Publishes attestations which block is not yet a known block.
+   *
+   * <p>These attestations should be passed to a wire module in order to request a block.
+   *
+   * @return a publisher.
+   */
+  Publisher<ReceivedAttestation> getUnknownAttestations();
+
+  /**
+   * Publishes aggregated attestations that are not yet included on chain.
+   *
+   * <p>It should be a source of attestations for block proposer.
+   *
+   * @return a publisher.
+   */
+  Publisher<OffChainAggregates> getAggregates();
+
+  /** Launches the pool. */
+  void start();
+
+  static AttestationPool create(
+      Publisher<ReceivedAttestation> source,
+      Publisher<SlotNumber> newSlots,
+      Publisher<Checkpoint> finalizedCheckpoints,
+      Publisher<BeaconBlock> importedBlocks,
+      Publisher<BeaconBlock> chainHeads,
+      Schedulers schedulers,
+      BeaconChainSpec spec,
+      BeaconChainStorage storage,
+      EmptySlotTransition emptySlotTransition) {
+
+    TimeFrameFilter timeFrameFilter = new TimeFrameFilter(spec, MAX_ATTESTATION_LOOKAHEAD);
+    SanityChecker sanityChecker = new SanityChecker(spec);
+    SignatureEncodingChecker encodingChecker = new SignatureEncodingChecker();
+    ProcessedAttestations processedFilter =
+        new ProcessedAttestations(spec::hash_tree_root, MAX_PROCESSED_ATTESTATIONS);
+    UnknownAttestationPool unknownAttestationPool =
+        new UnknownAttestationPool(
+            storage.getBlockStorage(), spec, MAX_ATTESTATION_LOOKAHEAD, MAX_UNKNOWN_ATTESTATIONS);
+    BatchVerifier batchVerifier =
+        new AttestationVerifier(storage.getTupleStorage(), spec, emptySlotTransition);
+
+    return new InMemoryAttestationPool(
+        source,
+        newSlots,
+        finalizedCheckpoints,
+        importedBlocks,
+        chainHeads,
+        schedulers,
+        timeFrameFilter,
+        sanityChecker,
+        encodingChecker,
+        processedFilter,
+        unknownAttestationPool,
+        batchVerifier);
+  }
+}

chain/src/main/java/org/ethereum/beacon/chain/pool/CheckedAttestation.java

@@ -0,0 +1,20 @@
+package org.ethereum.beacon.chain.pool;
+
+/** A simple DTO that carries an attestation and a result of some check run against it. */
+public class CheckedAttestation {
+  private final boolean passed;
+  private final ReceivedAttestation attestation;
+
+  public CheckedAttestation(boolean passed, ReceivedAttestation attestation) {
+    this.passed = passed;
+    this.attestation = attestation;
+  }
+
+  public boolean isPassed() {
+    return passed;
+  }
+
+  public ReceivedAttestation getAttestation() {
+    return attestation;
+  }
+}

chain/src/main/java/org/ethereum/beacon/chain/pool/InMemoryAttestationPool.java

@@ -0,0 +1,163 @@
+package org.ethereum.beacon.chain.pool;
+
+import java.util.List;
+import org.ethereum.beacon.chain.pool.checker.SanityChecker;
+import org.ethereum.beacon.chain.pool.checker.SignatureEncodingChecker;
+import org.ethereum.beacon.chain.pool.checker.TimeFrameFilter;
+import org.ethereum.beacon.chain.pool.churn.OffChainAggregates;
+import org.ethereum.beacon.chain.pool.reactor.ChurnProcessor;
+import org.ethereum.beacon.chain.pool.reactor.DoubleWorkProcessor;
+import org.ethereum.beacon.chain.pool.reactor.IdentificationProcessor;
+import org.ethereum.beacon.chain.pool.reactor.SanityProcessor;
+import org.ethereum.beacon.chain.pool.reactor.SignatureEncodingProcessor;
+import org.ethereum.beacon.chain.pool.reactor.TimeProcessor;
+import org.ethereum.beacon.chain.pool.reactor.VerificationProcessor;
+import org.ethereum.beacon.chain.pool.registry.ProcessedAttestations;
+import org.ethereum.beacon.chain.pool.registry.UnknownAttestationPool;
+import org.ethereum.beacon.chain.pool.verifier.BatchVerifier;
+import org.ethereum.beacon.core.BeaconBlock;
+import org.ethereum.beacon.core.state.Checkpoint;
+import org.ethereum.beacon.core.types.SlotNumber;
+import org.ethereum.beacon.schedulers.Schedulers;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.DirectProcessor;
+import reactor.core.publisher.Flux;
+import reactor.core.scheduler.Scheduler;
+
+/**
+ * An implementation of attestation pool based on <a href="https://projectreactor.io/">Reactor</a>
+ * library, one of the implementation of reactive streams.
+ */
+public class InMemoryAttestationPool implements AttestationPool {
+
+  private final Publisher<ReceivedAttestation> source;
+  private final Publisher<SlotNumber> newSlots;
+  private final Publisher<Checkpoint> finalizedCheckpoints;
+  private final Publisher<BeaconBlock> importedBlocks;
+  private final Publisher<BeaconBlock> chainHeads;
+  private final Schedulers schedulers;
+
+  private final TimeFrameFilter timeFrameFilter;
+  private final SanityChecker sanityChecker;
+  private final SignatureEncodingChecker encodingChecker;
+  private final ProcessedAttestations processedFilter;
+  private final UnknownAttestationPool unknownPool;
+  private final BatchVerifier verifier;
+
+  private final DirectProcessor<ReceivedAttestation> invalidAttestations = DirectProcessor.create();
+  private final DirectProcessor<ReceivedAttestation> validAttestations = DirectProcessor.create();
+  private final DirectProcessor<ReceivedAttestation> unknownAttestations = DirectProcessor.create();
+  private final DirectProcessor<OffChainAggregates> offChainAggregates = DirectProcessor.create();
+
+  public InMemoryAttestationPool(
+      Publisher<ReceivedAttestation> source,
+      Publisher<SlotNumber> newSlots,
+      Publisher<Checkpoint> finalizedCheckpoints,
+      Publisher<BeaconBlock> importedBlocks,
+      Publisher<BeaconBlock> chainHeads,
+      Schedulers schedulers,
+      TimeFrameFilter timeFrameFilter,
+      SanityChecker sanityChecker,
+      SignatureEncodingChecker encodingChecker,
+      ProcessedAttestations processedFilter,
+      UnknownAttestationPool unknownPool,
+      BatchVerifier batchVerifier) {
+    this.source = source;
+    this.newSlots = newSlots;
+    this.finalizedCheckpoints = finalizedCheckpoints;
+    this.importedBlocks = importedBlocks;
+    this.chainHeads = chainHeads;
+    this.schedulers = schedulers;
+    this.timeFrameFilter = timeFrameFilter;
+    this.sanityChecker = sanityChecker;
+    this.encodingChecker = encodingChecker;
+    this.processedFilter = processedFilter;
+    this.unknownPool = unknownPool;
+    this.verifier = batchVerifier;
+  }
+
+  @Override
+  public void start() {
+    Scheduler parallelExecutor =
+        schedulers
+            .newParallelDaemon("attestation-pool-%d", AttestationPool.MAX_THREADS)
+            .toReactor();
+
+    // create sources
+    Flux<ReceivedAttestation> sourceFx = Flux.from(source);
+    Flux<SlotNumber> newSlotsFx = Flux.from(newSlots);
+    Flux<Checkpoint> finalizedCheckpointsFx = Flux.from(finalizedCheckpoints);
+    Flux<BeaconBlock> importedBlocksFx = Flux.from(importedBlocks);
+    Flux<BeaconBlock> chainHeadsFx = Flux.from(chainHeads);
+
+    // check time frames
+    TimeProcessor timeProcessor =
+        new TimeProcessor(
+            timeFrameFilter, schedulers, sourceFx, finalizedCheckpointsFx, newSlotsFx);
+
+    // run sanity check
+    SanityProcessor sanityProcessor =
+        new SanityProcessor(sanityChecker, schedulers, timeProcessor, finalizedCheckpointsFx);
+
+    // discard already processed attestations
+    DoubleWorkProcessor doubleWorkProcessor =
+        new DoubleWorkProcessor(processedFilter, schedulers, sanityProcessor.getValid());
+
+    // check signature encoding
+    SignatureEncodingProcessor encodingProcessor =
+        new SignatureEncodingProcessor(
+            encodingChecker, doubleWorkProcessor.publishOn(parallelExecutor));
+
+    // identify attestation target
+    IdentificationProcessor identificationProcessor =
+        new IdentificationProcessor(
+            unknownPool, schedulers, encodingProcessor.getValid(), newSlotsFx, importedBlocksFx);
+
+    // verify attestations
+    Flux<List<ReceivedAttestation>> verificationThrottle =
+        identificationProcessor
+            .getIdentified()
+            .publishOn(parallelExecutor)
+            .bufferTimeout(VERIFIER_BUFFER_SIZE, VERIFIER_INTERVAL);
+    VerificationProcessor verificationProcessor =
+        new VerificationProcessor(verifier, verificationThrottle);
+
+    // feed churn
+    ChurnProcessor churnProcessor = new ChurnProcessor(schedulers, chainHeadsFx, newSlotsFx);
+
+    Scheduler outScheduler = schedulers.events().toReactor();
+    // expose valid attestations
+    verificationProcessor.getValid().publishOn(outScheduler).subscribe(validAttestations);
+    // expose not yet identified
+    identificationProcessor.getUnknown().publishOn(outScheduler).subscribe(unknownAttestations);
+    // expose aggregates
+    churnProcessor.publishOn(outScheduler).subscribe(offChainAggregates);
+    // expose invalid attestations
+    Flux.merge(
+            sanityProcessor.getInvalid(),
+            encodingProcessor.getInvalid(),
+            verificationProcessor.getInvalid())
+        .publishOn(outScheduler)
+        .subscribe(invalidAttestations);
+  }
+
+  @Override
+  public Publisher<ReceivedAttestation> getValid() {
+    return validAttestations;
+  }
+
+  @Override
+  public Publisher<ReceivedAttestation> getInvalid() {
+    return invalidAttestations;
+  }
+
+  @Override
+  public Publisher<ReceivedAttestation> getUnknownAttestations() {
+    return unknownAttestations;
+  }
+
+  @Override
+  public Publisher<OffChainAggregates> getAggregates() {
+    return offChainAggregates;
+  }
+}

chain/src/main/java/org/ethereum/beacon/chain/pool/ReceivedAttestation.java

@@ -0,0 +1,26 @@
+package org.ethereum.beacon.chain.pool;
+
+import org.ethereum.beacon.core.operations.Attestation;
+import org.ethereum.beacon.types.p2p.NodeId;
+
+/** An attestation received from the wire. */
+public class ReceivedAttestation {
+
+  /** An id of a node sent this attestation. */
+  private final NodeId sender;
+  /** An attestation message itself. */
+  private final Attestation message;
+
+  public ReceivedAttestation(NodeId sender, Attestation message) {
+    this.sender = sender;
+    this.message = message;
+  }
+
+  public NodeId getSender() {
+    return sender;
+  }
+
+  public Attestation getMessage() {
+    return message;
+  }
+}

chain/src/main/java/org/ethereum/beacon/chain/pool/StatefulProcessor.java

@@ -0,0 +1,24 @@
+package org.ethereum.beacon.chain.pool;
+
+/**
+ * Stateful processor.
+ *
+ * <p>A processor that requires particular inner state to be initialized before it could be safely
+ * called by its clients.
+ *
+ * <p>{@link #isInitialized()} method indicates whether processor has already been initialized or
+ * not. It's a client responsibility to check {@link #isInitialized()} result before calling to the
+ * instance of this processor.
+ *
+ * <p>Implementor MAY throw an {@link AssertionError} if it's been called before inner state has
+ * been initialised.
+ */
+public interface StatefulProcessor {
+
+  /**
+   * Checks whether processor state is initialized or not.
+   *
+   * @return {@code true} if processor is ready to work, {@link false}, otherwise.
+   */
+  boolean isInitialized();
+}