Merge pull request #1777 from ergoplatform/v4.0.36

Candidate for 4.0.36 release

README.md

@@ -82,7 +82,7 @@ To run specific Ergo version `<VERSION>` as a service with custom config `/path/
         -e MAX_HEAP=3G \
         ergoplatform/ergo:<VERSION> --<networkId> -c /etc/myergo.conf
 
-Available versions can be found on [Ergo Docker image page](https://hub.docker.com/r/ergoplatform/ergo/tags), for example, `v4.0.35`.
+Available versions can be found on [Ergo Docker image page](https://hub.docker.com/r/ergoplatform/ergo/tags), for example, `v4.0.36`.
 
 This will connect to the Ergo mainnet or testnet following your configuration passed in `myergo.conf` and network flag `--<networkId>`. Every default config value would be overwritten with corresponding value in `myergo.conf`. `MAX_HEAP` variable can be used to control how much memory can the node consume.
 

src/main/resources/api/openapi.yaml

@@ -1,7 +1,7 @@
 openapi: "3.0.2"
 
 info:
-  version: "4.0.35"
+  version: "4.0.36"
   title: Ergo Node API
   description: API docs for Ergo Node. Models are shared between all Ergo products
   contact:

src/main/resources/application.conf

@@ -391,7 +391,7 @@ scorex {
     nodeName = "ergo-node"
 
     # Network protocol version to be sent in handshakes
-    appVersion = 4.0.35
+    appVersion = 4.0.36
 
     # Network agent name. May contain information about client code
     # stack, starting from core code-base up to the end graphical interface.

src/main/resources/mainnet.conf

@@ -49,8 +49,8 @@ ergo {
     # The node still applying transactions to UTXO set and so checks UTXO set digests for each block.
     # Block at checkpoint height is to be checked against expected one.
     checkpoint = {
-      height = 798128
-      blockId = "e7f9066562fa123cff39d4f30f8e205591f55f6ae4ac802efa850fd54d5507f1"
+      height = 804900
+      blockId = "9d377e888dc7753ca5de3a7bc879afdc01993fd24f27bfe9c270d4828273c1a4"
     }
 
     # List with hex-encoded identifiers of transactions banned from getting into memory pool
@@ -65,7 +65,7 @@ scorex {
   network {
     magicBytes = [1, 0, 2, 4]
     bindAddress = "0.0.0.0:9030"
-    nodeName = "ergo-mainnet-4.0.35"
+    nodeName = "ergo-mainnet-4.0.36"
     nodeName = ${?NODENAME}
     knownPeers = [
       "213.239.193.208:9030",
@@ -81,10 +81,6 @@ scorex {
       "176.9.65.58:9130",
       "213.152.106.56:9030"
     ]
-
-    # Max number of delivery checks. Stop expecting modifier (and penalize peer) if it was not delivered after that
-    # number of delivery attempts
-    maxDeliveryChecks = 3
   }
   restApi {
 

src/main/resources/testnet.conf

@@ -77,7 +77,7 @@ scorex {
   network {
     magicBytes = [2, 0, 0, 2]
     bindAddress = "0.0.0.0:9020"
-    nodeName = "ergo-testnet-4.0.35"
+    nodeName = "ergo-testnet-4.0.36"
     nodeName = ${?NODENAME}
     knownPeers = [
       "213.239.193.208:9020",

src/main/scala/org/ergoplatform/http/api/ErgoBaseApiRoute.scala

@@ -1,20 +1,23 @@
 package org.ergoplatform.http.api
 
 import akka.actor.ActorRef
-import akka.http.scaladsl.server.{Directive1, ValidationRejection}
+import akka.http.scaladsl.server.{Directive1, Route, ValidationRejection}
 import org.ergoplatform.modifiers.mempool.ErgoTransaction
 import org.ergoplatform.nodeView.ErgoReadersHolder.{GetReaders, Readers}
 import org.ergoplatform.nodeView.mempool.ErgoMemPoolReader
 import org.ergoplatform.nodeView.state.{ErgoStateReader, UtxoStateReader}
 import org.ergoplatform.settings.{Algos, ErgoSettings}
-import scorex.core.api.http.ApiRoute
+import scorex.core.api.http.{ApiError, ApiRoute}
 import scorex.util.{ModifierId, bytesToId}
 import akka.pattern.ask
+import org.ergoplatform.nodeView.ErgoNodeViewHolder.ReceivableMessages.LocallyGeneratedTransaction
+import org.ergoplatform.nodeView.mempool.ErgoMemPool.ProcessingOutcome
+import org.ergoplatform.nodeView.mempool.ErgoMemPool.ProcessingOutcome._
 
 import scala.concurrent.{ExecutionContextExecutor, Future}
 import scala.util.{Success, Try}
 
-trait ErgoBaseApiRoute extends ApiRoute {
+trait ErgoBaseApiRoute extends ApiRoute with ApiCodecs {
 
   implicit val ec: ExecutionContextExecutor = context.dispatcher
 
@@ -36,6 +39,25 @@ trait ErgoBaseApiRoute extends ApiRoute {
     }
   }
 
+  /**
+    * Send local transaction to ErgoNodeViewHolder
+    * @return Transaction Id with status OK(200), or BadRequest(400)
+    */
+  protected def sendLocalTransactionRoute(nodeViewActorRef: ActorRef, tx: ErgoTransaction): Route = {
+    val resultFuture =
+      (nodeViewActorRef ? LocallyGeneratedTransaction(tx))
+        .mapTo[ProcessingOutcome]
+        .flatMap {
+          case Accepted => Future.successful(tx.id)
+          case DoubleSpendingLoser(_) => Future.failed(new IllegalArgumentException("Double spending attempt"))
+          case Declined(ex) => Future.failed(ex)
+          case Invalidated(ex) => Future.failed(ex)
+        }
+    completeOrRecoverWith(resultFuture) { ex =>
+      ApiError.BadRequest(ex.getMessage)
+    }
+  }
+
   /**
     * Helper method to verify transaction against UTXO set (and unconfirmed outputs in the mempool), or check
     * transaction syntax only if UTXO set is not available (the node is running in "digest" mode)

src/main/scala/org/ergoplatform/http/api/TransactionsApiRoute.scala

@@ -9,7 +9,6 @@ import org.ergoplatform.modifiers.mempool.ErgoTransaction
 import org.ergoplatform.nodeView.ErgoReadersHolder.{GetReaders, Readers}
 import org.ergoplatform.nodeView.mempool.ErgoMemPoolReader
 import org.ergoplatform.nodeView.mempool.HistogramStats.getFeeHistogram
-import org.ergoplatform.nodeView.ErgoNodeViewHolder.ReceivableMessages.LocallyGeneratedTransaction
 import org.ergoplatform.settings.ErgoSettings
 import scorex.core.api.http.ApiError.BadRequest
 import scorex.core.api.http.ApiResponse
@@ -41,7 +40,7 @@ case class TransactionsApiRoute(readersHolder: ActorRef,
     p.getAll.slice(offset, offset + limit).map(_.asJson).asJson
   }
 
-  private def validateTransactionAndProcess(tx: ErgoTransaction)(processFn: ErgoTransaction => Any): Route = {
+  private def validateTransactionAndProcess(tx: ErgoTransaction)(processFn: ErgoTransaction => Route): Route = {
     if (tx.size > ergoSettings.nodeSettings.maxTransactionSize) {
       BadRequest(s"Transaction $tx has too large size ${tx.size}")
     } else {
@@ -50,24 +49,19 @@ case class TransactionsApiRoute(readersHolder: ActorRef,
       } {
         _.fold(
           e => BadRequest(s"Malformed transaction: ${e.getMessage}"),
-          _ => {
-            processFn(tx)
-            ApiResponse(tx.id)
-          }
+          _ => processFn(tx)
         )
       }
     }
   }
 
 
   def sendTransactionR: Route = (pathEnd & post & entity(as[ErgoTransaction])) { tx =>
-    validateTransactionAndProcess(tx) { tx =>
-      nodeViewActorRef ! LocallyGeneratedTransaction(tx)
+      validateTransactionAndProcess(tx)(validTx => sendLocalTransactionRoute(nodeViewActorRef, validTx))
     }
-  }
 
   def checkTransactionR: Route = (path("check") & post & entity(as[ErgoTransaction])) { tx =>
-    validateTransactionAndProcess(tx) { tx => tx }
+    validateTransactionAndProcess(tx)(validTx => ApiResponse(validTx.id))
   }
 
   def getUnconfirmedTransactionsR: Route = (path("unconfirmed") & get & txPaging) { (offset, limit) =>

src/main/scala/org/ergoplatform/http/api/WalletApiRoute.scala

@@ -16,7 +16,6 @@ import org.ergoplatform.wallet.interface4j.SecretString
 import org.ergoplatform.wallet.Constants
 import org.ergoplatform.wallet.Constants.ScanId
 import org.ergoplatform.wallet.boxes.ErgoBoxSerializer
-import org.ergoplatform.nodeView.ErgoNodeViewHolder.ReceivableMessages.LocallyGeneratedTransaction
 import scorex.core.api.http.ApiError.{BadRequest, NotExists}
 import scorex.core.api.http.ApiResponse
 import scorex.core.settings.RESTApiSettings
@@ -158,10 +157,10 @@ case class WalletApiRoute(readersHolder: ActorRef,
                                             dataInputsRaw: Seq[String],
                                             verifyFn: ErgoTransaction => Future[Try[ErgoTransaction]],
                                             processFn: ErgoTransaction => Route): Route = {
-    withWalletOp(_.generateTransaction(requests, inputsRaw, dataInputsRaw).flatMap(txTry => txTry match {
+    withWalletOp(_.generateTransaction(requests, inputsRaw, dataInputsRaw).flatMap {
       case Success(tx) => verifyFn(tx)
       case f: Failure[ErgoTransaction] => Future(f)
-    })) {
+    }) {
       case Failure(e) => BadRequest(s"Bad request $requests. ${Option(e.getMessage).getOrElse(e.toString)}")
       case Success(tx) => processFn(tx)
     }
@@ -187,10 +186,8 @@ case class WalletApiRoute(readersHolder: ActorRef,
                               dataInputsRaw: Seq[String]): Route = {
     generateTransactionAndProcess(requests, inputsRaw, dataInputsRaw,
       tx => verifyTransaction(tx, readersHolder, ergoSettings),
-      { tx =>
-        nodeViewActorRef ! LocallyGeneratedTransaction(tx)
-        ApiResponse(tx.id)
-      })
+      validTx => sendLocalTransactionRoute(nodeViewActorRef, validTx)
+    )
   }
 
   def sendTransactionR: Route =

src/main/scala/org/ergoplatform/local/ErgoStatsCollector.scala

@@ -147,6 +147,8 @@ class ErgoStatsCollector(readersHolder: ActorRef,
       nodeInfo = nodeInfo.copy(
         stateRoot = Some(Algos.encode(fb.header.stateRoot)),
         stateVersion = Some(fb.encodedId))
+    case SemanticallySuccessfulModifier(_) =>
+      // Ignore other modifiers
   }
 
 }

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

@@ -25,7 +25,6 @@ import org.ergoplatform.nodeView.wallet.ErgoWalletReader
 import scorex.core.network.message.{InvSpec, MessageSpec, ModifiersSpec, RequestModifierSpec}
 import scorex.core.network._
 import scorex.core.network.message.{InvData, Message, ModifiersData}
-import scorex.core.network.{ConnectedPeer, ModifiersStatus, SendToPeer, SendToPeers}
 import scorex.core.serialization.ScorexSerializer
 import scorex.core.settings.NetworkSettings
 import scorex.core.transaction.Transaction
@@ -232,13 +231,7 @@ class ErgoNodeViewSynchronizer(networkControllerRef: ActorRef,
     * Processing sync V1 message `syncInfo` got from neighbour peer `remote`
     */
   protected def processSyncV1(hr: ErgoHistory, syncInfo: ErgoSyncInfoV1, remote: ConnectedPeer): Unit = {
-    val comparison = hr.compare(syncInfo)
-    log.debug(s"Comparison with $remote having starting points ${syncInfo.lastHeaderIds}. " +
-      s"Comparison result is $comparison.")
-
-    val oldStatus = syncTracker.getStatus(remote).getOrElse(Unknown)
-    val status = comparison
-    syncTracker.updateStatus(remote, status, height = None)
+    val (status, syncSendNeeded) = syncTracker.updateStatus(remote, syncInfo, hr)
 
     status match {
       case Unknown =>
@@ -279,7 +272,7 @@ class ErgoNodeViewSynchronizer(networkControllerRef: ActorRef,
         log.debug(s"$remote has equal header-chain")
     }
 
-    if ((oldStatus != status) || syncTracker.notSyncedOrOutdated(remote) || status == Older || status == Fork) {
+    if (syncSendNeeded) {
       val ownSyncInfo = getV1SyncInfo(hr)
       sendSyncToPeer(remote, ownSyncInfo)
     }
@@ -289,12 +282,7 @@ class ErgoNodeViewSynchronizer(networkControllerRef: ActorRef,
     * Processing sync V2 message `syncInfo` got from neighbour peer `remote` (supporting sync v2)
     */
   protected def processSyncV2(hr: ErgoHistory, syncInfo: ErgoSyncInfoV2, remote: ConnectedPeer): Unit = {
-    val oldStatus = syncTracker.getStatus(remote).getOrElse(Unknown)
-    val status = hr.compare(syncInfo)
-    syncTracker.updateStatus(remote, status, syncInfo.height)
-
-    log.debug(s"Comparison with $remote having starting points ${syncInfo.lastHeaders}. " +
-      s"Comparison result is $status.")
+    val (status, syncSendNeeded) = syncTracker.updateStatus(remote, syncInfo, hr)
 
     status match {
       case Unknown =>
@@ -330,7 +318,7 @@ class ErgoNodeViewSynchronizer(networkControllerRef: ActorRef,
         log.debug(s"$remote has equal header-chain")
     }
 
-    if ((oldStatus != status) || syncTracker.notSyncedOrOutdated(remote) || status == Older || status == Fork) {
+    if (syncSendNeeded) {
       val ownSyncInfo = getV2SyncInfo(hr, full = true)
       sendSyncToPeer(remote, ownSyncInfo)
     }
@@ -775,7 +763,7 @@ class ErgoNodeViewSynchronizer(networkControllerRef: ActorRef,
     networkControllerRef ! PenalizePeer(peer.connectionId.remoteAddress, PenaltyType.MisbehaviorPenalty)
   }
 
-  protected def penalizeMaliciousPeer(peer: ConnectedPeer): Unit = {
+  override protected def penalizeMaliciousPeer(peer: ConnectedPeer): Unit = {
     networkControllerRef ! PenalizePeer(peer.connectionId.remoteAddress, PenaltyType.PermanentPenalty)
   }
 

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

@@ -1,7 +1,7 @@
 package org.ergoplatform.network
 
 
-import org.ergoplatform.nodeView.history.ErgoHistory
+import org.ergoplatform.nodeView.history.{ErgoHistory, ErgoHistoryReader, ErgoSyncInfo, ErgoSyncInfoV1, ErgoSyncInfoV2}
 import org.ergoplatform.nodeView.history.ErgoHistory.Height
 import scorex.core.consensus.{Fork, PeerChainStatus, Older, Unknown}
 import scorex.core.network.ConnectedPeer
@@ -18,6 +18,11 @@ final case class ErgoSyncTracker(networkSettings: NetworkSettings, timeProvider:
   private val MinSyncInterval: FiniteDuration = 20.seconds
   private val SyncThreshold: FiniteDuration = 1.minute
 
+  /**
+    * After this timeout we clear peer's status
+    */
+  private val ClearThreshold: FiniteDuration = 3.minutes
+
   private[network] val statuses = mutable.Map[ConnectedPeer, ErgoPeerStatus]()
 
   def fullInfo(): Iterable[ErgoPeerStatus] = statuses.values
@@ -42,6 +47,28 @@ final case class ErgoSyncTracker(networkSettings: NetworkSettings, timeProvider:
     notSyncedOrMissing || outdated
   }
 
+  /**
+    * Obtains peer sync status from `syncInfo` network message and updates statuses table with it
+    *
+    * @return (new peer status, should our node send sync message to the peer)
+    */
+  def updateStatus(peer: ConnectedPeer,
+                   syncInfo: ErgoSyncInfo,
+                   hr: ErgoHistoryReader): (PeerChainStatus, Boolean) = {
+    val oldStatus = getStatus(peer).getOrElse(Unknown)
+    val status = hr.compare(syncInfo)
+
+    val height = syncInfo match {
+      case _: ErgoSyncInfoV1 => None
+      case sv2: ErgoSyncInfoV2 => sv2.height
+    }
+    updateStatus(peer, status, height)
+
+    val syncSendNeeded = (oldStatus != status) || notSyncedOrOutdated(peer) || status == Older || status == Fork
+
+    (status, syncSendNeeded)
+  }
+
   def updateStatus(peer: ConnectedPeer,
                    status: PeerChainStatus,
                    height: Option[Height]): Unit = {
@@ -55,7 +82,6 @@ final case class ErgoSyncTracker(networkSettings: NetworkSettings, timeProvider:
 
     val seniorsAfter = numOfSeniors()
 
-    // todo: we should also send NoBetterNeighbour signal when all the peers around are not seniors initially
     if (seniorsBefore > 0 && seniorsAfter == 0) {
       log.info("Syncing is done, switching to stable regime")
       // todo: update neighbours status ?
@@ -88,7 +114,22 @@ final case class ErgoSyncTracker(networkSettings: NetworkSettings, timeProvider:
     }
   }
 
-  protected[network] def outdatedPeers: IndexedSeq[ConnectedPeer] = {
+  /**
+    * Helper method to clear statuses of peers not updated for long enough
+    */
+  private[network] def clearOldStatuses(): Unit = {
+    val currentTime = timeProvider.time()
+    val peersToClear = statuses.filter { case (_, status) =>
+      status.lastSyncSentTime.exists(syncTime => (currentTime - syncTime).millis > ClearThreshold)
+    }.keys
+    if (peersToClear.nonEmpty) {
+      log.debug(s"Clearing stalled statuses for $peersToClear")
+      // we set status to `Unknown` and reset peer's height
+      peersToClear.foreach(p => updateStatus(p, Unknown, None))
+    }
+  }
+
+  private[network] def outdatedPeers: IndexedSeq[ConnectedPeer] = {
     val currentTime = timeProvider.time()
     statuses.filter { case (_, status) =>
       status.lastSyncSentTime.exists(syncTime => (currentTime - syncTime).millis > SyncThreshold)
@@ -121,6 +162,7 @@ final case class ErgoSyncTracker(networkSettings: NetworkSettings, timeProvider:
     * Updates lastSyncSentTime for all returned peers as a side effect
     */
   def peersToSyncWith(): IndexedSeq[ConnectedPeer] = {
+    clearOldStatuses()
     val outdated = outdatedPeers
     val peers =
       if (outdated.nonEmpty) {
@@ -130,8 +172,13 @@ final case class ErgoSyncTracker(networkSettings: NetworkSettings, timeProvider:
         val unknowns = statuses.filter(_._2.status == Unknown).toVector
         val forks = statuses.filter(_._2.status == Fork).toVector
         val elders = statuses.filter(_._2.status == Older).toVector
-        val nonOutdated =
-          (if (elders.nonEmpty) elders(scala.util.Random.nextInt(elders.size)) +: unknowns else unknowns) ++ forks
+
+        val eldersAndUnknown = if (elders.nonEmpty) {
+          elders(scala.util.Random.nextInt(elders.size)) +: unknowns
+        } else {
+          unknowns
+        }
+        val nonOutdated = eldersAndUnknown ++ forks
         nonOutdated.filter { case (_, status) =>
           (currentTime - status.lastSyncSentTime.getOrElse(0L)).millis >= MinSyncInterval
         }.map(_._1)