NODE-2033: Configurable UTX Pool fast lane (#3277)

Co-authored-by: Anvar Kiekbaev <[email protected]>

build.sbt

@@ -9,6 +9,7 @@
 import sbt.Keys._
 import sbt._
 import sbtcrossproject.CrossPlugin.autoImport.{CrossType, crossProject}
+import sbtcrossproject.CrossProject
 
 val langPublishSettings = Seq(
   coverageExcludedPackages := "",
@@ -31,9 +32,12 @@ lazy val lang =
       inConfig(Compile)(
         Seq(
           sourceGenerators += Tasks.docSource,
-          PB.targets += scalapb.gen(flatPackage = true) -> (sourceManaged).value,
+          PB.targets += scalapb.gen(flatPackage = true) -> sourceManaged.value,
           PB.protoSources := Seq(PB.externalIncludePath.value, baseDirectory.value.getParentFile / "shared" / "src" / "main" / "protobuf"),
-          includeFilter in PB.generate := new SimpleFileFilter((f: File) => f.getName == "DAppMeta.proto" || (f.getName.endsWith(".proto") && f.getParent.endsWith("waves"))),
+          includeFilter in PB.generate := { (f: File) =>
+            (** / "DAppMeta.proto").matches(f.toPath) ||
+              (** / "waves" / "*.proto").matches(f.toPath)
+          },
           PB.deleteTargetDirectory := false
         )
       )
@@ -152,6 +156,8 @@ packageAll := Def
       (node / Debian / packageBin).value
       (`grpc-server` / Universal / packageZipTarball).value
       (`grpc-server` / Debian / packageBin).value
+      (`blockchain-updates` / Universal / packageZipTarball).value
+      (`blockchain-updates` / Debian / packageBin).value
     }
   )
   .value

node-it/src/test/scala/com/wavesplatform/it/sync/UtxSuite.scala

@@ -8,27 +8,28 @@ import com.wavesplatform.it.Node
 import com.wavesplatform.it.api.SyncHttpApi._
 import com.wavesplatform.it.api.TransactionInfo
 import com.wavesplatform.it.transactions.NodesFromDocker
+import com.wavesplatform.lang.v1.estimator.ScriptEstimatorV1
 import com.wavesplatform.transaction.Asset.Waves
+import com.wavesplatform.transaction.TxVersion
+import com.wavesplatform.transaction.smart.script.ScriptCompiler
+import com.wavesplatform.transaction.smart.{InvokeScriptTransaction, SetScriptTransaction}
 import com.wavesplatform.transaction.transfer.TransferTransaction
 import org.scalatest.{CancelAfterFailure, FunSuite, Matchers}
 
 import scala.util.{Random, Try}
 
 class UtxSuite extends FunSuite with CancelAfterFailure with NodesFromDocker with Matchers {
-  override protected def nodeConfigs: Seq[Config] = UtxSuite.Configs
+  private val miner: Node    = nodes.head
+  private val notMiner: Node = nodes(1)
 
-  private def miner    = nodes.head
-  private def notMiner = nodes(1)
+  private var whitelistedAccount: KeyPair     = _
+  private var whitelistedDAppAccount: KeyPair = _
 
-  val ENOUGH_FEE = 5000000
-  val AMOUNT     = ENOUGH_FEE * 10
+  private val ENOUGH_FEE = 5000000
+  private val AMOUNT     = ENOUGH_FEE * 10
 
   test("Invalid transaction should be removed from from utx") {
-    val seed = Array.fill(32)(-1: Byte)
-
-    Random.nextBytes(seed)
-
-    val account = KeyPair(seed)
+    val account = UtxSuite.createAccount
 
     val transferToAccount = TransferTransaction
       .selfSigned(1.toByte, miner.keyPair, account.toAddress, Waves, AMOUNT, Waves, ENOUGH_FEE, ByteStr.empty, System.currentTimeMillis())
@@ -77,38 +78,163 @@ class UtxSuite extends FunSuite with CancelAfterFailure with NodesFromDocker wit
     assert(exactlyOneTxInBlockchain, "Only one tx should be in blockchain")
   }
 
+  test("Whitelisted transactions should be mined first of all") {
+    val minTransferFee  = 100000L
+    val minInvokeFee    = 500000L
+    val minSetScriptFee = 100000000L
+    val higherFee       = minInvokeFee * 2
+
+    val invokeAccount = UtxSuite.createAccount
+
+    def time: Long = System.currentTimeMillis()
+
+    val whitelistedAccountTransfer =
+      TransferTransaction
+        .selfSigned(
+          TxVersion.V1,
+          miner.keyPair,
+          whitelistedAccount.toAddress,
+          Waves,
+          5 * minTransferFee + 5,
+          Waves,
+          minTransferFee,
+          ByteStr.empty,
+          time
+        )
+        .explicitGet()
+    val whitelistedDAppAccountTransfer =
+      TransferTransaction
+        .selfSigned(
+          TxVersion.V1,
+          miner.keyPair,
+          whitelistedDAppAccount.toAddress,
+          Waves,
+          minSetScriptFee,
+          Waves,
+          minTransferFee,
+          ByteStr.empty,
+          time
+        )
+        .explicitGet()
+    val invokeAccountTransfer = TransferTransaction
+      .selfSigned(
+        TxVersion.V1,
+        miner.keyPair,
+        invokeAccount.toAddress,
+        Waves,
+        5 * minInvokeFee,
+        Waves,
+        minTransferFee,
+        ByteStr.empty,
+        time
+      )
+      .explicitGet()
+
+    Seq(whitelistedAccountTransfer, whitelistedDAppAccountTransfer, invokeAccountTransfer)
+      .map(tx => miner.signedBroadcast(tx.json()).id)
+      .foreach(nodes.waitForTransaction)
+
+    val scriptText =
+      """
+        |{-# STDLIB_VERSION 3 #-}
+        |{-# CONTENT_TYPE DAPP #-}
+        |{-# SCRIPT_TYPE ACCOUNT #-}
+        |@Callable(i)
+        |func default() = { WriteSet([DataEntry("0", true)]) }
+        |""".stripMargin
+    val script    = ScriptCompiler.compile(scriptText, ScriptEstimatorV1).explicitGet()._1
+    val setScript = SetScriptTransaction.selfSigned(TxVersion.V1, whitelistedDAppAccount, Some(script), minSetScriptFee, time).explicitGet()
+    miner.signedBroadcast(setScript.json())
+    nodes.waitForHeightAriseAndTxPresent(setScript.id().toString)
+
+    val txs = (1 to 5000).map { _ =>
+      TransferTransaction
+        .selfSigned(TxVersion.V1, miner.keyPair, UtxSuite.createAccount.toAddress, Waves, 1L, Waves, higherFee, ByteStr.empty, time)
+        .explicitGet()
+    }
+
+    val whitelistedTxs = {
+      val bySender = (1 to 5).map { _ =>
+        TransferTransaction
+          .selfSigned(TxVersion.V1, whitelistedAccount, UtxSuite.createAccount.toAddress, Waves, 1L, Waves, minTransferFee, ByteStr.empty, time)
+          .explicitGet()
+      }
+      val byDApp = (1 to 5).map { _ =>
+        InvokeScriptTransaction
+          .selfSigned(TxVersion.V1, invokeAccount, whitelistedDAppAccount.toAddress, None, Seq.empty, minInvokeFee, Waves, time)
+          .explicitGet()
+      }
+      Random.shuffle(bySender ++ byDApp)
+    }
+
+    txs.foreach(tx => miner.signedBroadcast(tx.json()))
+
+    miner.utxSize should be > 0
+
+    whitelistedTxs.map(tx => miner.signedBroadcast(tx.json()).id).foreach(nodes.waitForTransaction)
+
+    miner.utxSize should be > 0
+  }
+
   def txInBlockchain(txId: String, nodes: Seq[Node]): Boolean = {
     nodes.forall { node =>
       Try(node.transactionInfo[TransactionInfo](txId)).isSuccess
     }
   }
+
+  override protected def nodeConfigs: Seq[Config] = {
+    import UtxSuite._
+    import com.wavesplatform.it.NodeConfigs._
+
+    whitelistedAccount = createAccount
+    whitelistedDAppAccount = createAccount
+
+    val whitelist = Seq(whitelistedAccount, whitelistedDAppAccount).map(_.toAddress.stringRepr)
+
+    val minerConfig    = ConfigFactory.parseString(UtxSuite.minerConfigPredef(whitelist))
+    val notMinerConfig = ConfigFactory.parseString(UtxSuite.notMinerConfigPredef(whitelist))
+
+    Seq(
+      minerConfig.withFallback(Default.head),
+      notMinerConfig.withFallback(Default(1))
+    )
+  }
 }
 
 object UtxSuite {
-  import com.wavesplatform.it.NodeConfigs._
-  private val minerConfig = ConfigFactory.parseString(s"""
-                                                         |waves {
-                                                         |  synchronization.synchronization-timeout = 10s
-                                                         |  blockchain.custom.functionality {
-                                                         |    pre-activated-features.1 = 0
-                                                         |    generation-balance-depth-from-50-to-1000-after-height = 100
-                                                         |  }
-                                                         |  miner.quorum = 0
-                                                         |}""".stripMargin)
-
-  private val notMinerConfig = ConfigFactory.parseString(s"""
-                                                            |waves {
-                                                            |  synchronization.synchronization-timeout = 10s
-                                                            |  blockchain.custom.functionality {
-                                                            |    pre-activated-features.1 = 0
-                                                            |    generation-balance-depth-from-50-to-1000-after-height = 100
-                                                            |  }
-                                                            |  miner.enable = no
-                                                            |}""".stripMargin)
-
-  val Configs: Seq[Config] = Seq(
-    minerConfig.withFallback(Default.head),
-    notMinerConfig.withFallback(Default(1))
-  )
+  private def createAccount = {
+    val seed = Array.fill(32)(-1: Byte)
+    Random.nextBytes(seed)
+    KeyPair(seed)
+  }
 
+  private def minerConfigPredef(whitelist: Seq[String]) =
+    s"""
+       |waves {
+       |  synchronization.synchronization-timeout = 10s
+       |  utx {
+       |    max-size = 5000
+       |    fast-lane-addresses = [${whitelist.mkString(",")}]
+       |  }
+       |  blockchain.custom.functionality {
+       |    pre-activated-features.1 = 0
+       |    generation-balance-depth-from-50-to-1000-after-height = 100
+       |  }
+       |  miner.quorum = 0
+       |}""".stripMargin
+
+  private def notMinerConfigPredef(whitelist: Seq[String]) =
+    s"""
+       |waves {
+       |  synchronization.synchronization-timeout = 10s
+       |  utx {
+       |    max-size = 5000
+       |    fast-lane-addresses = [${whitelist.mkString(",")}]
+       |  }
+       |  blockchain.custom.functionality {
+       |    pre-activated-features.1 = 0
+       |    generation-balance-depth-from-50-to-1000-after-height = 100
+       |  }
+       |  miner.enable = no
+       |}""".stripMargin
 }

node/build.sbt

@@ -15,7 +15,6 @@ coverageExcludedPackages := ""
 
 inConfig(Compile)(
   Seq(
-    PB.protoSources in Compile := Seq(sourceDirectory.value / "protobuf"),
     PB.targets += scalapb.gen(flatPackage = true) -> sourceManaged.value,
     PB.deleteTargetDirectory := false,
     packageDoc / publishArtifact := false,
@@ -64,7 +63,8 @@ bashScriptExtraDefines ++= Seq(
   s"""addJava "-Dwaves.defaults.blockchain.type=${network.value}"""",
   s"""addJava "-Dwaves.defaults.directory=/var/lib/${(Universal / normalizedName).value}"""",
   s"""addJava "-Dwaves.defaults.config.directory=/etc/${(Universal / normalizedName).value}""""
-);
+)
+
 inConfig(Universal)(
   Seq(
     mappings += (baseDirectory.value / s"waves-sample.conf" -> "doc/waves.conf.sample"),

node/src/main/resources/application.conf

@@ -285,6 +285,8 @@ waves {
     blacklist-sender-addresses = []
     # Allow transfer transactions from the blacklisted addresses to these recipients (Base58 strings)
     allow-blacklisted-transfer-to = []
+    # Prioritize transactions from these addresses (Base58 strings)
+    fast-lane-addresses = []
     # Allow transactions from smart accounts
     allow-transactions-from-smart-accounts = true
     # Allow skipping checks with highest fee

node/src/main/scala/com/wavesplatform/consensus/TransactionsOrdering.scala

@@ -1,21 +1,24 @@
 package com.wavesplatform.consensus
 
 import com.wavesplatform.transaction.Asset.Waves
-import com.wavesplatform.transaction.Transaction
+import com.wavesplatform.transaction.smart.InvokeScriptTransaction
+import com.wavesplatform.transaction.{Authorized, Transaction}
 
 object TransactionsOrdering {
   trait WavesOrdering extends Ordering[Transaction] {
+    def isWhitelisted(t: Transaction): Boolean = false
     def txTimestampOrder(ts: Long): Long
-    private def orderBy(t: Transaction): (Double, Long, Long) = {
+    private def orderBy(t: Transaction): (Boolean, Double, Long, Long) = {
+      val byWhiteList = !isWhitelisted(t) // false < true
       val size        = t.bytes().length
       val byFee       = if (t.assetFee._1 != Waves) 0 else -t.assetFee._2
       val byTimestamp = txTimestampOrder(t.timestamp)
 
-      (byFee.toDouble / size.toDouble, byFee, byTimestamp)
+      (byWhiteList, byFee.toDouble / size.toDouble, byFee, byTimestamp)
     }
     override def compare(first: Transaction, second: Transaction): Int = {
       import Ordering.Double.TotalOrdering
-      implicitly[Ordering[(Double, Long, Long)]].compare(orderBy(first), orderBy(second))
+      implicitly[Ordering[(Boolean, Double, Long, Long)]].compare(orderBy(first), orderBy(second))
     }
   }
 
@@ -24,7 +27,14 @@ object TransactionsOrdering {
     override def txTimestampOrder(ts: Long): Long = -ts
   }
 
-  object InUTXPool extends WavesOrdering {
+  case class InUTXPool(whitelistAddresses: Set[String]) extends WavesOrdering {
+    override def isWhitelisted(t: Transaction): Boolean =
+      t match {
+        case _ if whitelistAddresses.isEmpty                                                            => false
+        case a: Authorized if whitelistAddresses.contains(a.sender.toAddress.stringRepr)                => true
+        case i: InvokeScriptTransaction if whitelistAddresses.contains(i.dAppAddressOrAlias.stringRepr) => true
+        case _                                                                                          => false
+      }
     override def txTimestampOrder(ts: Long): Long = ts
   }
 }

node/src/main/scala/com/wavesplatform/settings/UtxSettings.scala

@@ -6,6 +6,7 @@ case class UtxSettings(
     maxScriptedSize: Int,
     blacklistSenderAddresses: Set[String],
     allowBlacklistedTransferTo: Set[String],
+    fastLaneAddresses: Set[String],
     allowTransactionsFromSmartAccounts: Boolean,
     allowSkipChecks: Boolean
 )