input blocks data pruning

ergo-core/src/main/scala/org/ergoplatform/network/message/subblocks/InputBlockMessageSpec.scala → ergo-core/src/main/scala/org/ergoplatform/network/message/inputblocks/InputBlockMessageSpec.scala

@@ -1,4 +1,4 @@
-package org.ergoplatform.network.message.subblocks
+package org.ergoplatform.network.message.inputblocks
 
 import org.ergoplatform.network.message.MessageConstants.MessageCode
 import org.ergoplatform.network.message.MessageSpecInputBlocks

ergo-core/src/main/scala/org/ergoplatform/network/message/subblocks/InputBlockTransactionsData.scala → ergo-core/src/main/scala/org/ergoplatform/network/message/inputblocks/InputBlockTransactionsData.scala

@@ -1,4 +1,4 @@
-package org.ergoplatform.network.message.subblocks
+package org.ergoplatform.network.message.inputblocks
 
 import org.ergoplatform.modifiers.mempool.ErgoTransaction
 import scorex.util.ModifierId

ergo-core/src/main/scala/org/ergoplatform/network/message/subblocks/InputBlockTransactionsMessageSpec.scala → ergo-core/src/main/scala/org/ergoplatform/network/message/inputblocks/InputBlockTransactionsMessageSpec.scala

@@ -1,4 +1,4 @@
-package org.ergoplatform.network.message.subblocks
+package org.ergoplatform.network.message.inputblocks
 
 import org.ergoplatform.modifiers.mempool.ErgoTransactionSerializer
 import org.ergoplatform.network.message.MessageConstants.MessageCode

ergo-core/src/main/scala/org/ergoplatform/network/message/subblocks/InputBlockTransactionsRequestMessageSpec.scala → ergo-core/src/main/scala/org/ergoplatform/network/message/inputblocks/InputBlockTransactionsRequestMessageSpec.scala

@@ -1,4 +1,4 @@
-package org.ergoplatform.network.message.subblocks
+package org.ergoplatform.network.message.inputblocks
 
 import org.ergoplatform.network.message.MessageConstants.MessageCode
 import org.ergoplatform.network.message.MessageSpecInputBlocks

ergo-core/src/main/scala/org/ergoplatform/nodeView/LocallyGeneratedInputBlock.scala

@@ -1,6 +1,6 @@
 package org.ergoplatform.nodeView
 
-import org.ergoplatform.network.message.subblocks.InputBlockTransactionsData
+import org.ergoplatform.network.message.inputblocks.InputBlockTransactionsData
 import org.ergoplatform.subblocks.InputBlockInfo
 
 case class LocallyGeneratedInputBlock(sbi: InputBlockInfo, sbt: InputBlockTransactionsData)

src/main/scala/org/ergoplatform/mining/CandidateGenerator.scala

@@ -13,7 +13,7 @@ import org.ergoplatform.modifiers.history.header.{Header, HeaderWithoutPow}
 import org.ergoplatform.modifiers.history.popow.NipopowAlgos
 import org.ergoplatform.modifiers.mempool.{ErgoTransaction, UnconfirmedTransaction}
 import org.ergoplatform.network.ErgoNodeViewSynchronizerMessages._
-import org.ergoplatform.network.message.subblocks.InputBlockTransactionsData
+import org.ergoplatform.network.message.inputblocks.InputBlockTransactionsData
 import org.ergoplatform.nodeView.ErgoNodeViewHolder.ReceivableMessages.EliminateTransactions
 import org.ergoplatform.nodeView.ErgoReadersHolder.{GetReaders, Readers}
 import org.ergoplatform.nodeView.{LocallyGeneratedInputBlock, LocallyGeneratedOrderingBlock}
@@ -439,7 +439,7 @@ object CandidateGenerator extends ScorexLogging {
       val bestExtensionOpt: Option[Extension] = bestHeaderOpt
         .flatMap(h => history.typedModifierById[Extension](h.extensionId))
 
-      val lastSubblockOpt: Option[InputBlockInfo] = history.bestSubblock()
+      val lastSubblockOpt: Option[InputBlockInfo] = history.bestInputBlock()
 
       // there was sub-block generated before for this block
       val continueSubblock = lastSubblockOpt.exists(sbi => bestHeaderOpt.map(_.id).contains(sbi.header.parentId))

src/main/scala/org/ergoplatform/network/ErgoNodeViewSynchronizer.scala

@@ -33,7 +33,7 @@ import org.ergoplatform.consensus.{Equal, Fork, Nonsense, Older, Unknown, Younge
 import org.ergoplatform.modifiers.history.{ADProofs, ADProofsSerializer, BlockTransactions, BlockTransactionsSerializer}
 import org.ergoplatform.modifiers.history.extension.{Extension, ExtensionSerializer}
 import org.ergoplatform.modifiers.transaction.TooHighCostError
-import org.ergoplatform.network.message.subblocks.{InputBlockMessageSpec, InputBlockTransactionsData, InputBlockTransactionsMessageSpec, InputBlockTransactionsRequestMessageSpec}
+import org.ergoplatform.network.message.inputblocks.{InputBlockMessageSpec, InputBlockTransactionsData, InputBlockTransactionsMessageSpec, InputBlockTransactionsRequestMessageSpec}
 import org.ergoplatform.serialization.{ErgoSerializer, ManifestSerializer, SubtreeSerializer}
 import org.ergoplatform.subblocks.InputBlockInfo
 import scorex.crypto.authds.avltree.batch.VersionedLDBAVLStorage.splitDigest
@@ -1099,7 +1099,7 @@ class ErgoNodeViewSynchronizer(networkControllerRef: ActorRef,
   }
 
   def processInputBlockTransactionsRequest(subBlockId: ModifierId, hr: ErgoHistoryReader, remote: ConnectedPeer): Unit = {
-    hr.getSubBlockTransactions(subBlockId) match {
+    hr.getInputBlockTransactions(subBlockId) match {
       case Some(transactions) =>
         val std = InputBlockTransactionsData(subBlockId, transactions)
         val msg = Message(InputBlockTransactionsMessageSpec, Right(std), None)

src/main/scala/org/ergoplatform/network/ErgoNodeViewSynchronizerMessages.scala

@@ -11,7 +11,7 @@ import scorex.core.network.ConnectedPeer
 import scorex.util.ModifierId
 import org.ergoplatform.ErgoLikeContext.Height
 import org.ergoplatform.modifiers.history.popow.NipopowProof
-import org.ergoplatform.network.message.subblocks.InputBlockTransactionsData
+import org.ergoplatform.network.message.inputblocks.InputBlockTransactionsData
 import org.ergoplatform.subblocks.InputBlockInfo
 
 /**

src/main/scala/org/ergoplatform/nodeView/ErgoNodeViewHolder.scala

@@ -308,7 +308,7 @@ abstract class ErgoNodeViewHolder[State <: ErgoState[State]](settings: ErgoSetti
       history().applyInputBlock(sbi)
 
     case ProcessInputBlockTransactions(std) =>
-      history().applySubBlockTransactions(std.inputBlockID, std.transactions)
+      history().applyInputBlockTransactions(std.inputBlockID, std.transactions)
   }
 
   /**
@@ -687,7 +687,7 @@ abstract class ErgoNodeViewHolder[State <: ErgoState[State]](settings: ErgoSetti
     case LocallyGeneratedInputBlock(subblockInfo, subBlockTransactionsData) =>
       log.info(s"Got locally generated input block ${subblockInfo.header.id}")
       history().applyInputBlock(subblockInfo)
-      history().applySubBlockTransactions(subblockInfo.header.id, subBlockTransactionsData.transactions)
+      history().applyInputBlockTransactions(subblockInfo.header.id, subBlockTransactionsData.transactions)
       // todo: finish processing
   }
 

src/main/scala/org/ergoplatform/nodeView/history/storage/modifierprocessors/InputBlocksProcessor.scala

@@ -1,5 +1,6 @@
 package org.ergoplatform.nodeView.history.storage.modifierprocessors
 
+import org.ergoplatform.ErgoLikeContext.Height
 import org.ergoplatform.modifiers.mempool.ErgoTransaction
 import org.ergoplatform.subblocks.InputBlockInfo
 import scorex.util.{ModifierId, ScorexLogging, bytesToId}
@@ -24,45 +25,66 @@ trait InputBlocksProcessor extends ScorexLogging {
   // input block id -> input block transactions index
   val inputBlockTransactions = mutable.Map[ModifierId, Seq[ErgoTransaction]]()
 
+  private def bestInputBlockHeight: Option[Height] = _bestInputBlock.map(_.header.height)
+
+  private def prune() = {
+    val BlocksThreshold = 2 // we remove input-blocks data after 2 ordering blocks
+
+    val bestHeight = bestInputBlockHeight.getOrElse(0)
+    val idsToRemove = inputBlockRecords.flatMap{case (id, ibi) =>
+      val res = (bestHeight - ibi.header.height) > BlocksThreshold
+      if(res){
+        Some(id)
+      } else {
+        None
+      }
+    }
+    idsToRemove.foreach{ id =>
+      inputBlockRecords.remove(id)
+      inputBlockTransactions.remove(id)
+    }
+  }
+
   // reset sub-blocks structures, should be called on receiving ordering block (or slightly later?)
-  def resetState() = {
+  private def resetState() = {
     _bestInputBlock = None
-
-    // todo: subBlockRecords & subBlockTransactions should be cleared a bit later, as other peers may still ask for them
-    inputBlockRecords.clear()
-    inputBlockTransactions.clear()
+    prune()
   }
 
   // sub-blocks related logic
-  def applyInputBlock(sbi: InputBlockInfo): Unit = {
+  def applyInputBlock(ib: InputBlockInfo): Unit = {
     // new ordering block arrived ( should be processed outside ? )
-    if (sbi.header.height > _bestInputBlock.map(_.header.height).getOrElse(-1)) {
+    if (ib.header.height > _bestInputBlock.map(_.header.height).getOrElse(-1)) {
       resetState()
     }
 
-    inputBlockRecords.put(sbi.header.id, sbi)
+    inputBlockRecords.put(ib.header.id, ib)
 
     // todo: currently only one chain of subblocks considered,
     // todo: in fact there could be multiple trees here (one subblocks tree per header)
+    // todo: split best input header / block
     _bestInputBlock match {
-      case None => _bestInputBlock = Some(sbi)
-      case Some(maybeParent) if (sbi.prevInputBlockId.map(bytesToId).contains(maybeParent.header.id)) =>
-        _bestInputBlock = Some(sbi)
+      case None =>
+        log.debug(s"Applying best input block #: ${ib.header.id}, no parent")
+        _bestInputBlock = Some(ib)
+      case Some(maybeParent) if (ib.prevInputBlockId.map(bytesToId).contains(maybeParent.header.id)) =>
+        log.debug(s"Applying best input block #: ${ib.header.id}, parent is $maybeParent")
+        _bestInputBlock = Some(ib)
       case _ =>
         // todo: record it
-        log.debug(s"Applying non-best inpu block #: ${sbi.header.id}")
+        log.debug(s"Applying non-best input block #: ${ib.header.id}")
     }
   }
 
-  def applySubBlockTransactions(sbId: ModifierId, transactions: Seq[ErgoTransaction]): Unit = {
+  def applyInputBlockTransactions(sbId: ModifierId, transactions: Seq[ErgoTransaction]): Unit = {
     inputBlockTransactions.put(sbId, transactions)
   }
 
-  def getSubBlockTransactions(sbId: ModifierId): Option[Seq[ErgoTransaction]] = {
+  def getInputBlockTransactions(sbId: ModifierId): Option[Seq[ErgoTransaction]] = {
     inputBlockTransactions.get(sbId)
   }
 
-  def bestSubblock(): Option[InputBlockInfo] = {
+  def bestInputBlock(): Option[InputBlockInfo] = {
     _bestInputBlock
   }