Refactoring, fixes and additional tests

build.sbt

@@ -242,16 +242,16 @@ buildRIDERunnerForDocker := {
   )
 }
 
-lazy val compilePRRaw = taskKey[Unit]("Compile the project")
+lazy val compilePRRaw =
+  taskKey[Unit]("Incremental compilation (Compile + Test) without cleanup. Useful for fixing warnings/errors in a fast feedback loop.")
 compilePRRaw := Def
   .sequential(
-    clean.all(ScopeFilter(inAnyProject)),
     scalafmtCheck.all(ScopeFilter(inAnyProject, inConfigurations(Compile))),
     compile.all(ScopeFilter(inAnyProject, inConfigurations(Test)))
   )
   .value
 
-lazy val checkPRRaw = taskKey[Unit]("Compile the project and run unit tests")
+lazy val checkPRRaw = taskKey[Unit]("Incremental compilation without cleanup and running unit tests. Useful for quick test-fix iterations.")
 checkPRRaw := Def
   .sequential(
     compilePRRaw,
@@ -266,20 +266,28 @@ checkPRRaw := Def
   )
   .value
 
-def commandWithFatalWarnings(commandName: String, task: TaskKey[Unit]): Command =
-  Command.command(commandName) { state =>
+def commandWithCleanupAndFatalWarnings(commandName: String, task: TaskKey[Unit], help: Help): Command =
+  Command.command(commandName, help) { state =>
     val extracted = Project.extract(state)
     val newState = extracted.appendWithoutSession(
       Seq(Global / scalacOptions ++= Seq("-Werror")),
       state
     )
 
-    Project.extract(newState).runTask(task, newState)
+    Command.process("clean", newState, onParseError = _ => ()).unsafeRunTask(task)
     state
   }
 
-def compilePR: Command = commandWithFatalWarnings("compilePR", compilePRRaw)
-def checkPR: Command   = commandWithFatalWarnings("checkPR", checkPRRaw)
+def compilePR: Command = commandWithCleanupAndFatalWarnings(
+  "compilePR",
+  compilePRRaw,
+  Help.briefOnly(Seq("compilePR" -> "Compile with scalafmt."))
+)
+def checkPR: Command = commandWithCleanupAndFatalWarnings(
+  "checkPR",
+  checkPRRaw,
+  Help.briefOnly(Seq("checkPR" -> "Compile with scalafmt, build JavaScript artifacts, run unit tests."))
+)
 
 commands += Command.command("buildDebPackages") { state =>
   "set node / Debian / packageArchitecture := \"arm64\"" ::

node/src/main/scala/com/wavesplatform/api/common/CommonGeneratorsApi.scala

@@ -93,17 +93,22 @@ object CommonGeneratorsApi {
         addresses
           .lazyZip(txIds)
           .lazyZip(Iterator.from(0).take(addresses.size).map(GeneratorIndex(_)).to(Iterable))
-          .collect { case (Some(address), txnId, idx) => // TODO: address=None ?
-            val b = balances.get(idx) match {
-              case None if at.toInt <= blockchain.height => Some(0L)
-              case r                                     => r
-            }
+          .flatMap {
+            case (Some(address), txnId, idx) =>
+              val b = balances.get(idx) match {
+                case None if at.toInt <= blockchain.height => Some(0L)
+                case r                                     => r
+              }
+
+              Some(GeneratorEntry(address, b, txnId, conflict.heightOf(idx)))
 
-            GeneratorEntry(address, b, txnId, conflict.heightOf(idx))
+            case (None, txnId, idx) =>
+              log.warn(s"Can't find address, txnId=$txnId, idx=$idx. Contact with developers")
+              None
           }
           .toSeq
       } else {
-        log.warn(s"Different size: addresses=${addresses.size}, balances=${balances.size}, blsPks=${blsPks.size}")
+        log.warn(s"Different size: addresses=${addresses.size}, balances=${balances.size}, blsPks=${blsPks.size}. Contact with developers")
         Seq.empty
       }
     }

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

@@ -84,7 +84,6 @@ case class PoSSelector(blockchain: Blockchain, maxBaseTarget: Option[Long]) exte
   def validateGenerationSignature(block: Block): Either[ValidationError, ByteStr] = {
     val blockGenSig = block.header.generationSignature
 
-    // TODO: we already checked this
     blockchain.heightOf(block.header.reference).toRight(GenericError(s"Block reference ${block.header.reference} doesn't exist")).flatMap { height =>
       if (vrfActivated(height + 1)) {
         getHitSource(height)

node/src/main/scala/com/wavesplatform/database/Caches.scala

@@ -4,7 +4,7 @@ import com.google.common.cache.{CacheBuilder, CacheLoader, LoadingCache}
 import com.google.common.collect.ArrayListMultimap
 import com.google.protobuf.ByteString
 import com.typesafe.scalalogging.StrictLogging
-import com.wavesplatform.account.{Address, Alias, PublicKey}
+import com.wavesplatform.account.{Address, Alias}
 import com.wavesplatform.block.{Block, SignedBlockHeader}
 import com.wavesplatform.common.state.ByteStr
 import com.wavesplatform.common.utils.EitherExt2.*

node/src/main/scala/com/wavesplatform/database/RocksDBWriter.scala

@@ -7,7 +7,7 @@ import com.google.common.hash.{BloomFilter, Funnels}
 import com.google.common.primitives.Ints
 import com.google.common.util.concurrent.MoreExecutors
 import com.typesafe.scalalogging.Logger
-import com.wavesplatform.account.{Address, Alias, PublicKey}
+import com.wavesplatform.account.{Address, Alias}
 import com.wavesplatform.api.common.WavesBalanceIterator
 import com.wavesplatform.block.Block.BlockId
 import com.wavesplatform.block.BlockSnapshot
@@ -725,17 +725,13 @@ class RocksDBWriter(
       this.generationPeriodOf(h).foreach { currPeriod => // None checked in Caches
         if (nextCommittedGenerators.nonEmpty) {
           val nextPeriod = currPeriod.next
-
           rw.put(Keys.committedGenerators(nextPeriod, h), Some(nextCommittedGenerators))
-
-          // TODO: Option to not store
           rw.put(Keys.commitmentTransactions(nextPeriod, h), commitmentTransactionIds)
         }
 
         if (conflictGenerators.nonEmpty) rw.put(Keys.conflictGenerators(currPeriod, h), conflictGenerators)
       }
 
-      // TODO: Option to not store
       rw.put(Keys.generatorBalances(h, rdb.apiHandle), Some(generatorSet.map(x => x.index -> x.balance)))
 
       // TODO: height
@@ -1141,9 +1137,10 @@ class RocksDBWriter(
             rw.delete(Keys.transactionStateSnapshotAt(currentHeight, num, rdb.txSnapshotHandle))
           }
 
-          rw.delete(Keys.generatorBalances(currentHeight, rdb.apiHandle))
+          // Finality
           currentPeriod.foreach { currentPeriod =>
-            rw.delete(Keys.conflictGenerators(currentPeriod, currentHeight)) // TODO: test
+            rw.delete(Keys.generatorBalances(currentHeight, rdb.apiHandle))
+            rw.delete(Keys.conflictGenerators(currentPeriod, currentHeight))
 
             val nextPeriod = currentPeriod.next
             rw.delete(Keys.committedGenerators(nextPeriod, currentHeight))
@@ -1192,7 +1189,7 @@ class RocksDBWriter(
             Some(BlockSnapshot(block.id(), loadTxStateSnapshotsWithStatus(currentHeight, rdb, block.transactionData)))
           } else None
 
-          DiscardedBlock(block, Caches.toHitSource(discardedMeta), snapshot, Seq.empty) // TODO: generatorBalances
+          DiscardedBlock(block, Caches.toHitSource(discardedMeta), snapshot)
         }
 
         balancesToInvalidate.result().foreach(discardBalance)
@@ -1643,7 +1640,7 @@ class RocksDBWriter(
   override def resolveERC20Address(address: ERC20Address): Option[IssuedAsset] =
     readOnly(_.get(Keys.assetStaticInfo(address)).map(assetInfo => IssuedAsset(assetInfo.id.toByteStr)))
 
-  override def lastStateHash(refId: Option[ByteStr]): ByteStr =
+  override def lastStateHash(liquidBlockId: Option[ByteStr]): ByteStr =
     snapshotStateHash(height)
 
   def snapshotStateHash(height: Int): ByteStr =

node/src/main/scala/com/wavesplatform/mining/Miner.scala

@@ -204,7 +204,7 @@ class MinerImpl(
         }
       } yield balance
 
-      def retryReasons(balance: Long) = for {
+      def retryReasons(balance: Long): Either[String, ForgeAttemptResult] = for {
         _ <- checkQuorumAvailable()
         validBlockDelay <- pos
           .getValidBlockDelay(height, account, refBaseTarget, balance)
@@ -317,18 +317,20 @@ class MinerImpl(
           f"Next attempt for acc=${account.toAddress} in ${offset.toUnit(SECONDS)}%.3f seconds (${LocalTime.now().plusNanos(offset.toNanos)})$waitBlockIdStr"
         )
 
+        // We need to wait, because the mining scheduled for SnapshotBlockchain state in BlockchainUpdater
+        // If we need to forge a block immediately, we can't do this, because the parent block is not in the state yet
         val waitBlockAppendedTask = waitBlockId match {
           case Some(blockId) =>
             def waitUntilBlockAppended(block: BlockId): Task[Unit] =
-              if (blockchainUpdater.contains(block)) Task.unit
-              else Task.defer(waitUntilBlockAppended(block)).delayExecution(1 seconds)
+              if (blockchainUpdater.lastBlockId.contains(block)) Task.unit
+              else Task.defer(waitUntilBlockAppended(block)).delayExecution(1.second)
 
             waitUntilBlockAppended(blockId)
 
           case None => Task.unit
         }
 
-        def appendTask(block: Block, totalConstraint: MiningConstraint) = // TODO: accept blockAppender instead all these dependencies?
+        def appendTask(block: Block, totalConstraint: MiningConstraint) =
           BlockAppender(blockchainUpdater, timeService, utx, pos, blockEndorser, appenderScheduler)(block, None).flatMap {
             case Left(BlockFromFuture(_, _)) => // Time was corrected, retry
               generateBlockTask(account, None)

node/src/main/scala/com/wavesplatform/network/PeerDatabase.scala

@@ -46,6 +46,6 @@ object PeerDatabase {
 
     override val detailedSuspended: Map[InetAddress, Long] = Map.empty
 
-    override def blacklistAndClose(channel: Channel, reason: String): Unit = channel.close()
+    override def blacklistAndClose(channel: Channel, reason: String): Unit = Option(channel).foreach(_.close())
   }
 }

node/src/main/scala/com/wavesplatform/network/RxExtensionLoader.scala

@@ -22,6 +22,8 @@ import monix.reactive.{Observable, Observer}
 import java.util.concurrent.TimeUnit
 import scala.concurrent.duration.*
 
+/** @param blocks The recent is last
+  */
 case class ExtensionBlocks(remoteScore: BigInt, blocks: Seq[Block], snapshots: Map[BlockId, BlockSnapshotResponse]) {
   override def toString: String = s"ExtensionBlocks($remoteScore, ${formatSignatures(blocks.map(_.id()))}"
 }

node/src/main/scala/com/wavesplatform/state/Blockchain.scala

@@ -92,14 +92,15 @@ trait Blockchain {
 
   def effectiveBalanceBanHeights(address: Address): Seq[Int]
 
-  // TODO: named?
   def committedGenerators(at: GenerationPeriod): IndexedSeq[(Address, BlsPublicKey)]
 
   def conflictGenerators(at: GenerationPeriod): ConflictGenerators
 
   def resolveERC20Address(address: ERC20Address): Option[IssuedAsset]
 
-  def lastStateHash(refId: Option[ByteStr]): ByteStr
+  /** @note Works only for key block or one of liquid blocks
+    */
+  def lastStateHash(liquidBlockId: Option[ByteStr]): ByteStr
 }
 
 object Blockchain {
@@ -174,9 +175,6 @@ object Blockchain {
       generationDeposit = blockchain.generationDeposit(address)
     )
 
-    // TODO: lock?
-    // TODO: not efficient? See RocksDBWriter.balanceSnapshots
-    // TODO: optimize
     def generationDeposit(address: Address, at: Height = Height(blockchain.height)): Long = blockchain.generationPeriodOf(at).fold(0L) { period =>
       val committed = blockchain.committedGenerators(period)
       val conflict  = blockchain.conflictGenerators(period)