NODE-2609 Light node implementation

benchmark/src/main/scala/com/wavesplatform/state/DBState.scala

@@ -26,7 +26,8 @@ abstract class DBState extends ScorexLogging {
     new RocksDBWriter(
       rdb,
       settings.blockchainSettings,
-      settings.dbSettings.copy(maxCacheSize = 1)
+      settings.dbSettings.copy(maxCacheSize = 1),
+      settings.enableLightMode
     )
 
   AddressScheme.current = new AddressScheme { override val chainId: Byte = 'W' }

benchmark/src/test/scala/com/wavesplatform/RollbackBenchmark.scala

@@ -22,7 +22,7 @@ object RollbackBenchmark extends ScorexLogging {
     val settings      = Application.loadApplicationConfig(Some(new File(args(0))))
     val rdb           = RDB.open(settings.dbSettings)
     val time          = new NTP(settings.ntpServer)
-    val rocksDBWriter = new RocksDBWriter(rdb, settings.blockchainSettings, settings.dbSettings)
+    val rocksDBWriter = new RocksDBWriter(rdb, settings.blockchainSettings, settings.dbSettings, settings.enableLightMode)
 
     val issuer = KeyPair(new Array[Byte](32))
 
@@ -80,6 +80,7 @@ object RollbackBenchmark extends ScorexLogging {
       0,
       None,
       genesisBlock.header.generationSignature,
+      ByteStr.empty,
       genesisBlock
     )
 
@@ -103,7 +104,7 @@ object RollbackBenchmark extends ScorexLogging {
     val nextSnapshot = StateSnapshot.build(rocksDBWriter, portfolios2.toMap).explicitGet()
 
     log.info("Appending next block")
-    rocksDBWriter.append(nextSnapshot, 0, 0, None, ByteStr.empty, nextBlock)
+    rocksDBWriter.append(nextSnapshot, 0, 0, None, ByteStr.empty, ByteStr.empty, nextBlock)
 
     log.info("Rolling back")
     val start = System.nanoTime()

benchmark/src/test/scala/com/wavesplatform/state/BaseState.scala

@@ -2,7 +2,6 @@ package com.wavesplatform.state
 
 import java.io.File
 import java.nio.file.Files
-
 import com.typesafe.config.ConfigFactory
 import com.wavesplatform.account.KeyPair
 import com.wavesplatform.block.Block
@@ -76,11 +75,12 @@ trait BaseState {
     .block
 
   private def append(prev: Option[Block], next: Block): Unit = {
-    val preconditionSnapshot =
-      BlockDiffer.fromBlock(state, prev, next, MiningConstraint.Unlimited, next.header.generationSignature)
+    val differResult =
+      BlockDiffer
+        .fromBlock(state, prev, next, None, MiningConstraint.Unlimited, next.header.generationSignature)
         .explicitGet()
-        .snapshot
-    state.append(preconditionSnapshot, 0, 0, None, next.header.generationSignature, next)
+
+    state.append(differResult.snapshot, 0, 0, None, next.header.generationSignature, differResult.computedStateHash, next)
   }
 
   def applyBlock(b: Block): Unit = {

benchmark/src/test/scala/com/wavesplatform/state/RocksDBWriterBenchmark.scala

@@ -87,7 +87,7 @@ object RocksDBWriterBenchmark {
       RDB.open(wavesSettings.dbSettings)
     }
 
-    val db = new RocksDBWriter(rawDB, wavesSettings.blockchainSettings, wavesSettings.dbSettings)
+    val db = new RocksDBWriter(rawDB, wavesSettings.blockchainSettings, wavesSettings.dbSettings, wavesSettings.enableLightMode)
 
     def loadBlockInfoAt(height: Int): Option[(BlockMeta, Seq[(TxMeta, Transaction)])] =
       loadBlockMetaAt(height).map { meta =>

benchmark/src/test/scala/com/wavesplatform/state/WavesEnvironmentBenchmark.scala

@@ -136,7 +136,7 @@ object WavesEnvironmentBenchmark {
     }
 
     val environment: Environment[Id] = {
-      val state = new RocksDBWriter(rdb, wavesSettings.blockchainSettings, wavesSettings.dbSettings)
+      val state = new RocksDBWriter(rdb, wavesSettings.blockchainSettings, wavesSettings.dbSettings, wavesSettings.enableLightMode)
       new WavesEnvironment(
         AddressScheme.current.chainId,
         Coeval.raiseError(new NotImplementedError("`tx` is not implemented")),

grpc-server/src/test/scala/com/wavesplatform/events/BlockchainUpdatesSpec.scala

@@ -860,6 +860,7 @@ class BlockchainUpdatesSpec extends FreeSpec with WithBUDomain with ScalaFutures
           xtnBuybackAddress,
           d.blockchain
         )
+
         append.stateUpdate.get.balances shouldBe Seq(
           PBBalanceUpdate(
             challengingMiner.toAddress.toByteString,

node-it/src/test/scala/com/wavesplatform/it/BaseTargetChecker.scala

@@ -38,7 +38,7 @@ object BaseTargetChecker {
             blockchainUpdater.isFeatureActivated(BlockchainFeatures.TransactionStateSnapshot)
           )
           .explicitGet()
-      blockchainUpdater.processBlock(genesisBlock, genesisBlock.header.generationSignature)
+      blockchainUpdater.processBlock(genesisBlock, genesisBlock.header.generationSignature, None)
 
       NodeConfigs.Default.map(_.withFallback(sharedConfig)).collect {
         case cfg if cfg.as[Boolean]("waves.miner.enable") =>

node-it/src/test/scala/com/wavesplatform/it/NodeConfigs.scala

@@ -73,6 +73,8 @@ object NodeConfigs {
       s"waves.blockchain.custom.functionality.min-asset-info-update-interval = $blocks"
 
     val nonMiner: String = "waves.miner.enable = no"
+
+    val lightNode: String = "waves.enable-light-mode = true"
   }
 
 }

node-it/src/test/scala/com/wavesplatform/it/api/model.scala

@@ -5,8 +5,8 @@ import com.wavesplatform.common.state.ByteStr
 import com.wavesplatform.state.DataEntry
 import com.wavesplatform.transaction.assets.exchange.AssetPair
 import com.wavesplatform.transaction.transfer.MassTransferTransaction.Transfer
-import io.grpc.{Metadata, Status => GrpcStatus}
-import play.api.libs.json._
+import io.grpc.{Metadata, Status as GrpcStatus}
+import play.api.libs.json.*
 
 import scala.util.{Failure, Success}
 
@@ -799,10 +799,11 @@ case class Block(
     reward: Option[Long],
     desiredReward: Option[Long],
     vrf: Option[String],
+    challengedHeader: Option[ChallengedBlockHeader],
     version: Option[Byte] = None
 )
 object Block {
-  import PublicKey._
+  import PublicKey.*
 
   implicit val blockFormat: Format[Block] = Format(
     Reads(jsv =>
@@ -827,6 +828,7 @@ object Block {
         baseTarget          <- (jsv \ "nxt-consensus" \ "base-target").validateOpt[Int]
         transactionsRoot    <- (jsv \ "transactionsRoot").validateOpt[String]
         vrf                 <- (jsv \ "VRF").validateOpt[String]
+        challengedHeader    <- (jsv \ "challengedHeader").validateOpt[ChallengedBlockHeader]
       } yield Block(
         id,
         signature,
@@ -847,6 +849,7 @@ object Block {
         reward,
         desiredReward,
         vrf,
+        challengedHeader,
         version
       )
     ),
@@ -870,6 +873,7 @@ case class BlockHeader(
     desiredReward: Option[Long],
     totalFee: Long,
     vrf: Option[String],
+    challengedHeader: Option[ChallengedBlockHeader],
     version: Option[Byte] = None
 )
 object BlockHeader {
@@ -892,6 +896,7 @@ object BlockHeader {
         baseTarget          <- (jsv \ "nxt-consensus" \ "base-target").validateOpt[Int]
         transactionsRoot    <- (jsv \ "transactionsRoot").validateOpt[String]
         vrf                 <- (jsv \ "VRF").validateOpt[String]
+        challengedHeader    <- (jsv \ "challengedHeader").validateOpt[ChallengedBlockHeader]
       } yield BlockHeader(
         id,
         signature,
@@ -908,13 +913,26 @@ object BlockHeader {
         desiredReward,
         totalFee,
         vrf,
+        challengedHeader,
         version
       )
     ),
     Json.writes[BlockHeader]
   )
 }
 
+case class ChallengedBlockHeader(
+    headerSignature: String,
+    features: Set[Short],
+    generator: String,
+    generatorPublicKey: String,
+    desiredReward: Long,
+    stateHash: Option[String]
+)
+object ChallengedBlockHeader {
+  implicit val challengedBlockHeaderFormat: Format[ChallengedBlockHeader] = Json.format
+}
+
 case class GenerationSignatureResponse(generationSignature: String)
 object GenerationSignatureResponse {
   implicit val generationSignatureResponseFormat: Format[GenerationSignatureResponse] = Json.format

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

@@ -0,0 +1,136 @@
+package com.wavesplatform.it.sync
+
+import com.typesafe.config.Config
+import com.wavesplatform.account.{AddressScheme, KeyPair}
+import com.wavesplatform.block.Block
+import com.wavesplatform.common.state.ByteStr
+import com.wavesplatform.common.utils.EitherExt2
+import com.wavesplatform.consensus.FairPoSCalculator
+import com.wavesplatform.crypto
+import com.wavesplatform.features.BlockchainFeatures
+import com.wavesplatform.it.api.Block as ApiBlock
+import com.wavesplatform.it.{BaseFunSuite, NodeConfigs, TransferSending}
+import com.wavesplatform.it.api.AsyncNetworkApi.NodeAsyncNetworkApi
+import com.wavesplatform.it.api.SyncHttpApi.*
+import com.wavesplatform.network.RawBytes
+import com.wavesplatform.transaction.Asset.Waves
+import com.wavesplatform.transaction.{Proofs, Transaction, TxPositiveAmount}
+import com.wavesplatform.transaction.transfer.TransferTransaction
+
+import scala.concurrent.Await
+import scala.concurrent.duration.*
+
+class BlockChallengeSuite extends BaseFunSuite with TransferSending {
+  override def nodeConfigs: Seq[Config] =
+    NodeConfigs.newBuilder
+      .overrideBase(_.quorum(4))
+      .overrideBase(_.preactivatedFeatures(BlockchainFeatures.TransactionStateSnapshot.id.toInt -> 0))
+      .withDefault(1)
+      .withSpecial(1, _.lightNode)
+      .withSpecial(2, _.nonMiner)
+      .buildNonConflicting()
+
+  test("NODE-1167. All nodes should receive and apply block with challenge") {
+    val challenger = nodes.head
+    val malicious  = nodes.last
+
+    val height    = challenger.height
+    val lastBlock = challenger.blockAt(height)
+
+    val txs = (1 to 3).map { idx =>
+      TransferTransaction(
+        3.toByte,
+        challenger.publicKey,
+        malicious.keyPair.toAddress,
+        Waves,
+        TxPositiveAmount.unsafeFrom(100000000 * (idx + 1)),
+        Waves,
+        TxPositiveAmount.unsafeFrom(1000000),
+        ByteStr.empty,
+        System.currentTimeMillis(),
+        Proofs.empty,
+        AddressScheme.current.chainId
+      ).signWith(challenger.keyPair.privateKey)
+    }
+    val invalidBlock = createBlockWithInvalidStateHash(lastBlock, height, malicious.keyPair, txs)
+    waitForBlockTime(invalidBlock)
+    Await.ready(challenger.sendByNetwork(RawBytes.fromBlock(invalidBlock)), 2.minutes)
+
+    txs.foreach { tx =>
+      nodes.waitForTransaction(tx.id().toString)
+    }
+
+    val challengingIds = nodes.map { node =>
+      val challengingBlock = node.blockAt(height + 1)
+      checkChallengingBlock(challengingBlock, invalidBlock, challenger.address, txs)
+      challengingBlock.id
+    }
+
+    challengingIds.toSet.size shouldBe 1
+  }
+
+  private def checkChallengingBlock(challengingBlock: ApiBlock, challengedBlock: Block, challengerAddress: String, txs: Seq[Transaction]) = {
+    challengingBlock.challengedHeader shouldBe defined
+    val challengedHeader = challengingBlock.challengedHeader.get
+    challengedHeader.headerSignature shouldBe challengedBlock.signature.toString
+    challengedHeader.features shouldBe challengedBlock.header.featureVotes.toSet
+    challengedHeader.desiredReward shouldBe challengedBlock.header.rewardVote
+    challengedHeader.stateHash shouldBe challengedBlock.header.stateHash.map(_.toString)
+    challengedHeader.generator shouldBe challengedBlock.header.generator.toAddress.toString
+    challengedHeader.generatorPublicKey shouldBe challengedBlock.header.generator.toString
+    challengingBlock.generator shouldBe challengerAddress
+    challengingBlock.transactions.map(_.id).toSet shouldBe txs.map(_.id().toString).toSet
+  }
+
+  private def createBlockWithInvalidStateHash(lastBlock: ApiBlock, height: Int, signer: KeyPair, txs: Seq[Transaction]): Block = {
+    val lastBlockVrfOrGenSig = lastBlock.vrf.orElse(lastBlock.generationSignature).map(str => ByteStr.decodeBase58(str).get).get.arr
+    val genSig: ByteStr      = crypto.signVRF(signer.privateKey, lastBlockVrfOrGenSig)
+
+    val hitSource =
+      crypto.verifyVRF(genSig, lastBlockVrfOrGenSig, signer.publicKey).explicitGet()
+
+    val posCalculator = FairPoSCalculator.V1
+    val version       = 5.toByte
+
+    val validBlockDelay: Long = posCalculator
+      .calculateDelay(
+        hit(hitSource.arr),
+        lastBlock.baseTarget.get,
+        nodes.head.accountBalances(signer.toAddress.toString)._2
+      )
+
+    val baseTarget: Long = posCalculator
+      .calculateBaseTarget(
+        10,
+        height,
+        lastBlock.baseTarget.get,
+        lastBlock.timestamp,
+        None,
+        lastBlock.timestamp + validBlockDelay
+      )
+
+    Block
+      .buildAndSign(
+        version = version,
+        timestamp = lastBlock.timestamp + validBlockDelay,
+        reference = ByteStr.decodeBase58(lastBlock.id).get,
+        baseTarget = baseTarget,
+        generationSignature = genSig,
+        txs = txs,
+        signer = signer,
+        featureVotes = Seq(22),
+        rewardVote = 1000000000L,
+        stateHash = Some(ByteStr.fill(32)(1)),
+        challengedHeader = None
+      )
+      .explicitGet()
+  }
+
+  private def hit(generatorSignature: Array[Byte]): BigInt = BigInt(1, generatorSignature.take(8).reverse)
+
+  private def waitForBlockTime(block: Block): Unit = {
+    val timeout = block.header.timestamp - System.currentTimeMillis()
+
+    if (timeout > 0) Thread.sleep(timeout)
+  }
+}

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

@@ -0,0 +1,45 @@
+package com.wavesplatform.it.sync
+
+import com.google.common.primitives.Ints
+import com.typesafe.config.Config
+import com.wavesplatform.account.{Address, PublicKey}
+import com.wavesplatform.it.{BaseFunSuite, NodeConfigs, TransferSending}
+import com.wavesplatform.it.api.SyncHttpApi.*
+
+class LightNodeBroadcastSuite extends BaseFunSuite with TransferSending {
+  override def nodeConfigs: Seq[Config] =
+    NodeConfigs.newBuilder
+      .overrideBase(_.quorum(0))
+      .overrideBase(_.preactivatedFeatures((14, 1000000)))
+      .withDefault(1)
+      .withSpecial(2, _.lightNode)
+      .buildNonConflicting()
+
+  test("NODE-1162. Light node should correctly broadcast blocks and snapshots to the other light nodes") {
+    val fullNode   = nodes.head
+    val lightNode1 = nodes(1)
+    val lightNode2 = nodes(2)
+
+    fullNode.blacklist(lightNode2.networkAddress)
+    lightNode2.blacklist(fullNode.networkAddress)
+
+    (1 to 3).foreach { blockIdx =>
+      val transactionIds = (1 to 50).map { idx =>
+        val destPk   = Ints.toByteArray(blockIdx) ++ Ints.toByteArray(idx) ++ new Array[Byte](24)
+        val destAddr = Address.fromPublicKey(PublicKey(destPk)).toString
+        sender.transfer(sender.keyPair, destAddr, idx).id
+      }
+      transactionIds.foreach(nodes.waitForTransaction(_))
+      nodes.waitForHeightArise()
+    }
+
+    val height = nodes.waitForHeightArise()
+
+    val fullNodeState   = fullNode.debugStateAt(height - 1)
+    val lightNodeState1 = lightNode1.debugStateAt(height - 1)
+    val lightNodeState2 = lightNode2.debugStateAt(height - 1)
+
+    fullNodeState.toSet shouldBe lightNodeState1.toSet
+    lightNodeState1.toSet shouldBe lightNodeState2.toSet
+  }
+}