Feature: Add solidification stage to pipeline and only broadcast solid transactions

src/main/java/com/iota/iri/Iota.java

@@ -29,6 +29,8 @@
 import com.iota.iri.service.spentaddresses.SpentAddressesService;
 import com.iota.iri.service.tipselection.TipSelector;
 import com.iota.iri.service.transactionpruning.TransactionPruner;
+import com.iota.iri.service.validation.TransactionSolidifier;
+import com.iota.iri.service.validation.TransactionValidator;
 import com.iota.iri.storage.Indexable;
 import com.iota.iri.storage.Persistable;
 import com.iota.iri.storage.PersistenceProvider;
@@ -98,6 +100,8 @@ public class Iota {
 
     public final MilestoneSolidifier milestoneSolidifier;
 
+    public final TransactionSolidifier transactionSolidifier;
+
     public final BundleValidator bundleValidator;
 
     public final Tangle tangle;
@@ -116,7 +120,7 @@ public class Iota {
      * @param configuration Information about how this node will be configured.
      *
      */
-    public Iota(IotaConfig configuration, SpentAddressesProvider spentAddressesProvider, SpentAddressesService spentAddressesService, SnapshotProvider snapshotProvider, SnapshotService snapshotService, LocalSnapshotManager localSnapshotManager, MilestoneService milestoneService, LatestMilestoneTracker latestMilestoneTracker, LatestSolidMilestoneTracker latestSolidMilestoneTracker, SeenMilestonesRetriever seenMilestonesRetriever, LedgerService ledgerService, TransactionPruner transactionPruner, MilestoneSolidifier milestoneSolidifier, BundleValidator bundleValidator, Tangle tangle, TransactionValidator transactionValidator, TransactionRequester transactionRequester, NeighborRouter neighborRouter, TransactionProcessingPipeline transactionProcessingPipeline, TipsRequester tipsRequester, TipsViewModel tipsViewModel, TipSelector tipsSelector) {
+    public Iota(IotaConfig configuration, SpentAddressesProvider spentAddressesProvider, SpentAddressesService spentAddressesService, SnapshotProvider snapshotProvider, SnapshotService snapshotService, LocalSnapshotManager localSnapshotManager, MilestoneService milestoneService, LatestMilestoneTracker latestMilestoneTracker, LatestSolidMilestoneTracker latestSolidMilestoneTracker, SeenMilestonesRetriever seenMilestonesRetriever, LedgerService ledgerService, TransactionPruner transactionPruner, MilestoneSolidifier milestoneSolidifier, BundleValidator bundleValidator, Tangle tangle, TransactionValidator transactionValidator, TransactionRequester transactionRequester, NeighborRouter neighborRouter, TransactionProcessingPipeline transactionProcessingPipeline, TipsRequester tipsRequester, TipsViewModel tipsViewModel, TipSelector tipsSelector, TransactionSolidifier transactionSolidifier) {
         this.configuration = configuration;
 
         this.ledgerService = ledgerService;
@@ -134,6 +138,7 @@ public Iota(IotaConfig configuration, SpentAddressesProvider spentAddressesProvi
         this.neighborRouter = neighborRouter;
         this.txPipeline = transactionProcessingPipeline;
         this.tipsRequester = tipsRequester;
+        this.transactionSolidifier = transactionSolidifier;
 
         // legacy classes
         this.bundleValidator = bundleValidator;
@@ -185,8 +190,6 @@ public void init() throws Exception {
             tangle.clearMetadata(com.iota.iri.model.persistables.Transaction.class);
         }
 
-        transactionValidator.init();
-
         txPipeline.start();
         neighborRouter.start();
         tipsRequester.start();
@@ -195,6 +198,7 @@ public void init() throws Exception {
         latestSolidMilestoneTracker.start();
         seenMilestonesRetriever.start();
         milestoneSolidifier.start();
+        transactionSolidifier.start();
 
         if (localSnapshotManager != null) {
             localSnapshotManager.start(latestMilestoneTracker);
@@ -236,6 +240,7 @@ private void rescanDb() throws Exception {
     public void shutdown() throws Exception {
         // shutdown in reverse starting order (to not break any dependencies)
         milestoneSolidifier.shutdown();
+        transactionSolidifier.shutdown();
         seenMilestonesRetriever.shutdown();
         latestSolidMilestoneTracker.shutdown();
         latestMilestoneTracker.shutdown();
@@ -250,7 +255,6 @@ public void shutdown() throws Exception {
         tipsRequester.shutdown();
         txPipeline.shutdown();
         neighborRouter.shutdown();
-        transactionValidator.shutdown();
         tangle.shutdown();
 
         // free the resources of the snapshot provider last because all other instances need it

src/main/java/com/iota/iri/MainInjectionConfiguration.java

@@ -50,6 +50,9 @@
 import com.iota.iri.service.tipselection.impl.WalkerAlpha;
 import com.iota.iri.service.transactionpruning.TransactionPruner;
 import com.iota.iri.service.transactionpruning.async.AsyncTransactionPruner;
+import com.iota.iri.service.validation.TransactionSolidifier;
+import com.iota.iri.service.validation.TransactionValidator;
+import com.iota.iri.service.validation.impl.TransactionSolidifierImpl;
 import com.iota.iri.storage.Persistable;
 import com.iota.iri.storage.PersistenceProvider;
 import com.iota.iri.storage.Tangle;
@@ -137,8 +140,8 @@ SeenMilestonesRetriever provideSeenMilestonesRetriever(Tangle tangle, SnapshotPr
 
     @Singleton
     @Provides
-    MilestoneSolidifier provideMilestoneSolidifier(SnapshotProvider snapshotProvider, TransactionValidator transactionValidator) {
-        return new MilestoneSolidifierImpl(snapshotProvider, transactionValidator);
+    MilestoneSolidifier provideMilestoneSolidifier(SnapshotProvider snapshotProvider, TransactionSolidifier transactionSolidifier) {
+        return new MilestoneSolidifierImpl(snapshotProvider, transactionSolidifier);
     }
 
     @Singleton
@@ -163,6 +166,12 @@ TransactionValidator provideTransactionValidator(Tangle tangle, SnapshotProvider
         return new TransactionValidator(tangle, snapshotProvider, tipsViewModel, transactionRequester, configuration);
     }
 
+    @Singleton
+    @Provides
+    TransactionSolidifier provideTransactionSolidifier(Tangle tangle, SnapshotProvider snapshotProvider, TransactionRequester transactionRequester){
+        return new TransactionSolidifierImpl(tangle, snapshotProvider, transactionRequester);
+    }
+
     @Singleton
     @Provides
     TipSelector provideTipSelector(Tangle tangle, SnapshotProvider snapshotProvider,
@@ -178,8 +187,8 @@ TipSelector provideTipSelector(Tangle tangle, SnapshotProvider snapshotProvider,
 
     @Singleton
     @Provides
-    Iota provideIota(SpentAddressesProvider spentAddressesProvider, SpentAddressesService spentAddressesService, SnapshotProvider snapshotProvider, SnapshotService snapshotService, @Nullable LocalSnapshotManager localSnapshotManager, MilestoneService milestoneService, LatestMilestoneTracker latestMilestoneTracker, LatestSolidMilestoneTracker latestSolidMilestoneTracker, SeenMilestonesRetriever seenMilestonesRetriever, LedgerService ledgerService, @Nullable TransactionPruner transactionPruner, MilestoneSolidifier milestoneSolidifier, BundleValidator bundleValidator, Tangle tangle, TransactionValidator transactionValidator, TransactionRequester transactionRequester, NeighborRouter neighborRouter, TransactionProcessingPipeline transactionProcessingPipeline, TipsRequester tipsRequester, TipsViewModel tipsViewModel, TipSelector tipsSelector) {
-        return new Iota(configuration, spentAddressesProvider, spentAddressesService, snapshotProvider, snapshotService, localSnapshotManager, milestoneService, latestMilestoneTracker, latestSolidMilestoneTracker, seenMilestonesRetriever, ledgerService, transactionPruner, milestoneSolidifier, bundleValidator, tangle, transactionValidator, transactionRequester, neighborRouter, transactionProcessingPipeline, tipsRequester, tipsViewModel, tipsSelector);
+    Iota provideIota(SpentAddressesProvider spentAddressesProvider, SpentAddressesService spentAddressesService, SnapshotProvider snapshotProvider, SnapshotService snapshotService, @Nullable LocalSnapshotManager localSnapshotManager, MilestoneService milestoneService, LatestMilestoneTracker latestMilestoneTracker, LatestSolidMilestoneTracker latestSolidMilestoneTracker, SeenMilestonesRetriever seenMilestonesRetriever, LedgerService ledgerService, @Nullable TransactionPruner transactionPruner, MilestoneSolidifier milestoneSolidifier, BundleValidator bundleValidator, Tangle tangle, TransactionValidator transactionValidator, TransactionRequester transactionRequester, NeighborRouter neighborRouter, TransactionProcessingPipeline transactionProcessingPipeline, TipsRequester tipsRequester, TipsViewModel tipsViewModel, TipSelector tipsSelector, TransactionSolidifier transactionSolidifier) {
+        return new Iota(configuration, spentAddressesProvider, spentAddressesService, snapshotProvider, snapshotService, localSnapshotManager, milestoneService, latestMilestoneTracker, latestSolidMilestoneTracker, seenMilestonesRetriever, ledgerService, transactionPruner, milestoneSolidifier, bundleValidator, tangle, transactionValidator, transactionRequester, neighborRouter, transactionProcessingPipeline, tipsRequester, tipsViewModel, tipsSelector, transactionSolidifier);
     }
 
     @Singleton

src/main/java/com/iota/iri/controllers/TransactionViewModel.java

@@ -3,6 +3,7 @@
 import com.iota.iri.model.*;
 import com.iota.iri.model.persistables.*;
 import com.iota.iri.service.snapshot.Snapshot;
+import com.iota.iri.service.validation.TransactionValidator;
 import com.iota.iri.storage.Indexable;
 import com.iota.iri.storage.Persistable;
 import com.iota.iri.storage.Tangle;
@@ -712,26 +713,11 @@ public void setMetadata() {
                 : TransactionViewModel.FILLED_SLOT;
     }
 
-    public static void updateSolidTransactions(Tangle tangle, Snapshot initialSnapshot, final LinkedHashSet<Hash> analyzedHashes)
-            throws Exception {
-        Object[] hashes = analyzedHashes.toArray();
-        TransactionViewModel transactionViewModel;
-        for(int i = hashes.length -1; i >= 0; i--){
-            transactionViewModel = TransactionViewModel.fromHash(tangle, (Hash) hashes[i]);
-
-            transactionViewModel.updateHeights(tangle, initialSnapshot);
-
-            if (!transactionViewModel.isSolid()) {
-                transactionViewModel.updateSolid(true);
-                transactionViewModel.update(tangle, initialSnapshot, "solid|height");
-            }
-        }
-    }
 
     /**
      * Updates the {@link Transaction#solid} value of the referenced {@link Transaction} object.
      *
-     * Used by the {@link com.iota.iri.TransactionValidator} to quickly set the solidity of a {@link Transaction} set.
+     * Used by the {@link TransactionValidator} to quickly set the solidity of a {@link Transaction} set.
      *
      * @param solid The solidity of the transaction in the database
      * @return True if the {@link Transaction#solid} has been updated, False if not.

src/main/java/com/iota/iri/network/NetworkInjectionConfiguration.java

@@ -3,7 +3,8 @@
 import com.google.inject.AbstractModule;
 import com.google.inject.Provides;
 import com.google.inject.Singleton;
-import com.iota.iri.TransactionValidator;
+import com.iota.iri.service.validation.TransactionSolidifier;
+import com.iota.iri.service.validation.TransactionValidator;
 import com.iota.iri.conf.IotaConfig;
 import com.iota.iri.controllers.TipsViewModel;
 import com.iota.iri.network.impl.TipsRequesterImpl;
@@ -46,9 +47,9 @@ TipsRequester provideTipsRequester(NeighborRouter neighborRouter, Tangle tangle,
     TransactionProcessingPipeline provideTransactionProcessingPipeline(NeighborRouter neighborRouter,
             TransactionValidator txValidator, Tangle tangle, SnapshotProvider snapshotProvider,
             TipsViewModel tipsViewModel, LatestMilestoneTracker latestMilestoneTracker,
-            TransactionRequester transactionRequester) {
+            TransactionRequester transactionRequester, TransactionSolidifier transactionSolidifier) {
         return new TransactionProcessingPipelineImpl(neighborRouter, configuration, txValidator, tangle,
-                snapshotProvider, tipsViewModel, latestMilestoneTracker, transactionRequester);
+                snapshotProvider, tipsViewModel, latestMilestoneTracker, transactionRequester, transactionSolidifier);
     }
 
     @Singleton

src/main/java/com/iota/iri/network/pipeline/ReceivedStage.java

@@ -1,6 +1,6 @@
 package com.iota.iri.network.pipeline;
 
-import com.iota.iri.TransactionValidator;
+import com.iota.iri.service.validation.TransactionValidator;
 import com.iota.iri.controllers.TransactionViewModel;
 import com.iota.iri.network.TransactionRequester;
 import com.iota.iri.network.neighbor.Neighbor;
@@ -91,8 +91,8 @@ public ProcessingContext process(ProcessingContext ctx) {
         }
 
         // broadcast the newly saved tx to the other neighbors
-        ctx.setNextStage(TransactionProcessingPipeline.Stage.BROADCAST);
-        ctx.setPayload(new BroadcastPayload(originNeighbor, tvm));
+        ctx.setNextStage(TransactionProcessingPipeline.Stage.SOLIDIFY);
+        ctx.setPayload(new SolidifyPayload(originNeighbor, tvm));
         return ctx;
     }
 }

src/main/java/com/iota/iri/network/pipeline/SolidifyPayload.java

@@ -0,0 +1,39 @@
+package com.iota.iri.network.pipeline;
+
+import com.iota.iri.controllers.TransactionViewModel;
+import com.iota.iri.network.neighbor.Neighbor;
+
+/**
+ * Defines a payload which gets submitted to the {@link SolidifyStage}.
+ */
+public class SolidifyPayload extends Payload {
+    private Neighbor originNeighbor;
+    private TransactionViewModel tvm;
+
+    /**
+     * Constructor for solidification payload.
+     *
+     * @param originNeighbor    The originating point of a received transaction
+     * @param tvm               The transaction that needs to be solidified
+     */
+    public SolidifyPayload(Neighbor originNeighbor, TransactionViewModel tvm){
+        this.originNeighbor = originNeighbor;
+        this.tvm = tvm;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public Neighbor getOriginNeighbor(){
+        return originNeighbor;
+    }
+
+    /**
+     * Fetches the transaction from the payload.
+     * @return          The transaction stored in the payload.
+     */
+    public TransactionViewModel getTransaction(){
+        return tvm;
+    }
+}

src/main/java/com/iota/iri/network/pipeline/SolidifyStage.java

@@ -0,0 +1,108 @@
+package com.iota.iri.network.pipeline;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.iota.iri.controllers.TipsViewModel;
+import com.iota.iri.controllers.TransactionViewModel;
+import com.iota.iri.model.Hash;
+import com.iota.iri.service.validation.TransactionSolidifier;
+import com.iota.iri.service.validation.TransactionValidator;
+import com.iota.iri.storage.Tangle;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static com.iota.iri.controllers.TransactionViewModel.fromHash;
+
+/**
+ * The {@link SolidifyStage} is used to process newly received transaction for solidity. Once a transaction has been
+ * passed from the {@link ReceivedStage} it will be placed into this stage to have the {@link TransactionSolidifier}
+ * check the solidity of the transaction. If the transaction is found to be solid, it will be passed forward to the
+ * {@link BroadcastStage}. If it is found to be unsolid, it is put through the solidity check so that missing reference
+ * transactions get requested. If the transaction is unsolid, a random solid tip is broadcast instead to keep the
+ * requests transmitting to neighbors.
+ */
+public class SolidifyStage implements Stage {
+    private static final Logger log = LoggerFactory.getLogger(SolidifyStage.class);
+
+    private TransactionSolidifier txSolidifier;
+    private TransactionValidator txValidator;
+    private TipsViewModel tipsViewModel;
+    private Tangle tangle;
+    private TransactionViewModel tip;
+
+    /**
+     * Constructor for the {@link SolidifyStage}.
+     *
+     * @param txSolidifier      Transaction solidifier implementation for updating transaction's solidity status
+     * @param txValidator       Transaction validator implementation for determining the validity of a transaction
+     * @param tipsViewModel     Used for broadcasting random solid tips if the subject transaction is unsolid
+     * @param tangle            A reference to the nodes DB
+     */
+    public SolidifyStage(TransactionSolidifier txSolidifier, TransactionValidator txValidator,
+                         TipsViewModel tipsViewModel, Tangle tangle){
+        this.txSolidifier = txSolidifier;
+        this.txValidator = txValidator;
+        this.tipsViewModel = tipsViewModel;
+        this.tangle = tangle;
+    }
+
+    /**
+     * Processes the payload of the {@link ProcessingContext} as a {@link SolidifyPayload}. First the transaction will
+     * be checked for solidity and validity. If the transaction is already solid or can be set solid quickly by the
+     * transaction validator, the transaction is passed to the {@link BroadcastStage}. If not, the transaction is
+     * added to the solidification queue, and a random solid tip is pulled form the {@link TipsViewModel} to be
+     * broadcast instead.
+     *
+     * @param ctx       The context to process
+     * @return          The output context, in most cases a {@link BroadcastPayload}.
+     */
+    @Override
+    public ProcessingContext process(ProcessingContext ctx){
+        try {
+            SolidifyPayload payload = (SolidifyPayload) ctx.getPayload();
+            TransactionViewModel tvm = payload.getTransaction();
+
+            if (tvm.isSolid() || txValidator.quickSetSolid(tvm)) {
+                ctx.setNextStage(TransactionProcessingPipeline.Stage.BROADCAST);
+                ctx.setPayload(new BroadcastPayload(payload.getOriginNeighbor(), payload.getTransaction()));
+                return ctx;
+            }
+
+            txSolidifier.addToSolidificationQueue(tvm.getHash());
+
+            return broadcastTip(ctx, payload);
+        }catch (Exception e){
+            log.error("Failed to process transaction for solidification", e);
+            ctx.setNextStage(TransactionProcessingPipeline.Stage.ABORT);
+            return ctx;
+        }
+
+    }
+
+    private ProcessingContext broadcastTip(ProcessingContext ctx, SolidifyPayload payload) throws  Exception{
+        if(tip == null) {
+            Hash tipHash = tipsViewModel.getRandomSolidTipHash();
+
+            if (tipHash == null) {
+                ctx.setNextStage(TransactionProcessingPipeline.Stage.FINISH);
+                return ctx;
+            }
+
+            tip = fromHash(tangle, tipHash);
+        }
+
+        if(tip.isSolid()) {
+            ctx.setNextStage(TransactionProcessingPipeline.Stage.BROADCAST);
+            ctx.setPayload(new BroadcastPayload(payload.getOriginNeighbor(), tip));
+            return ctx;
+        }
+
+        ctx.setNextStage(TransactionProcessingPipeline.Stage.FINISH);
+        return ctx;
+    }
+
+    @VisibleForTesting
+    void injectTip(TransactionViewModel tvm) throws Exception {
+        tip = tvm;
+        tip.updateSolid(true);
+    }
+}

src/main/java/com/iota/iri/network/pipeline/TransactionProcessingPipeline.java

@@ -1,6 +1,7 @@
 package com.iota.iri.network.pipeline;
 
 import com.iota.iri.network.neighbor.Neighbor;
+import com.iota.iri.service.validation.TransactionSolidifier;
 
 import java.nio.ByteBuffer;
 import java.util.concurrent.BlockingQueue;
@@ -14,7 +15,7 @@ public interface TransactionProcessingPipeline {
      * Defines the different stages of the {@link TransactionProcessingPipelineImpl}.
      */
     enum Stage {
-        PRE_PROCESS, HASHING, VALIDATION, REPLY, RECEIVED, BROADCAST, MULTIPLE, ABORT, FINISH,
+        PRE_PROCESS, HASHING, VALIDATION, REPLY, RECEIVED, BROADCAST, MULTIPLE, ABORT, FINISH, SOLIDIFY,
     }
 
     /**
@@ -65,6 +66,12 @@ enum Stage {
      */
     void process(byte[] txTrits);
 
+    /**
+     * Fetches a set of transactions from the {@link TransactionSolidifier} and submits
+     * the object into the {@link BroadcastStage} queue.
+     */
+    void refillBroadcastQueue();
+
     /**
      * Shut downs the pipeline by shutting down all stages.
      */
@@ -111,4 +118,11 @@ enum Stage {
      * @param hashingStage the {@link HashingStage} to use
      */
     void setHashingStage(HashingStage hashingStage);
+
+    /**
+     * Sets the solidify stage. This method should only be used for injecting mocked objects.
+     *
+     * @param solidifyStage the {@link SolidifyStage} to use
+     */
+    void setSolidifyStage(SolidifyStage solidifyStage);
 }