NODE-2618 Fixed GRPC getNFTList (#3904)

grpc-server/src/test/scala/com/wavesplatform/api/grpc/test/AssetsApiGrpcSpec.scala

@@ -3,6 +3,7 @@ package com.wavesplatform.api.grpc.test
 import com.google.protobuf.ByteString
 import com.wavesplatform.account.KeyPair
 import com.wavesplatform.api.grpc.{AssetInfoResponse, AssetsApiGrpcImpl, NFTRequest, NFTResponse}
+import com.wavesplatform.block.Block.ProtoBlockVersion
 import com.wavesplatform.db.WithDomain
 import com.wavesplatform.db.WithState.AddrWithBalance
 import com.wavesplatform.features.BlockchainFeatures
@@ -55,6 +56,49 @@ class AssetsApiGrpcSpec extends FreeSpec with BeforeAndAfterAll with DiffMatcher
     }
   }
 
+  "NODE-999. GetNftList limit should work properly" in withDomain(
+    RideV6.addFeatures(BlockchainFeatures.ReduceNFTFee),
+    AddrWithBalance.enoughBalances(sender)
+  ) { d =>
+    val nftIssues = (1 to 5).map(idx => TxHelpers.issue(sender, 1, name = s"nft$idx", reissuable = false))
+    val limit     = 2
+    val afterId   = 1 // second element
+
+    d.appendBlock()
+    val mb1 = d.appendMicroBlock(nftIssues.take(afterId + 1)*)
+    d.appendMicroBlock(nftIssues.drop(afterId + 1)*)
+
+    // full liquid
+    d.rocksDBWriter.containsTransaction(nftIssues(afterId)) shouldBe false
+    d.rocksDBWriter.containsTransaction(nftIssues(afterId + 1)) shouldBe false
+    check()
+
+    // liquid afterId
+    d.appendBlock(d.createBlock(ProtoBlockVersion, nftIssues.drop(afterId + 1), Some(mb1)))
+    d.rocksDBWriter.containsTransaction(nftIssues(afterId)) shouldBe true
+    d.rocksDBWriter.containsTransaction(nftIssues(afterId + 1)) shouldBe false
+    check()
+
+    // full solid
+    d.appendBlock()
+    d.rocksDBWriter.containsTransaction(nftIssues(afterId)) shouldBe true
+    d.rocksDBWriter.containsTransaction(nftIssues(afterId + 1)) shouldBe true
+    check()
+
+    def check() = {
+      val (observer, result) = createObserver[NFTResponse]
+      val request = NFTRequest.of(
+        ByteString.copyFrom(sender.toAddress.bytes),
+        limit,
+        afterAssetId = ByteString.copyFrom(nftIssues(afterId).asset.id.arr)
+      )
+      getGrpcApi(d).getNFTList(request, observer)
+      val response = result.runSyncUnsafe()
+      response.size shouldBe limit
+      response.map(_.assetInfo.get.name) shouldBe nftIssues.slice(afterId + 1, afterId + limit + 1).map(_.name.toStringUtf8)
+    }
+  }
+
   private def getGrpcApi(d: Domain) =
     new AssetsApiGrpcImpl(d.assetsApi, d.accountsApi)
 }

grpc-server/src/test/scala/com/wavesplatform/api/grpc/test/GRPCBroadcastSpec.scala

@@ -1,7 +1,7 @@
 package com.wavesplatform.api.grpc.test
 
 import scala.concurrent.{Await, Future}
-import scala.concurrent.duration.Duration
+import scala.concurrent.duration.*
 import com.wavesplatform.account.Address
 import com.wavesplatform.common.state.ByteStr
 import com.wavesplatform.test.{FlatSpec, TestTime}
@@ -91,7 +91,7 @@ class GRPCBroadcastSpec extends FlatSpec with BeforeAndAfterAll with PathMockFac
 
     @throws[StatusException]("on failed broadcast")
     def assertBroadcast(tx: Transaction): Unit = {
-      Await.result(grpcTxApi.broadcast(PBTransactions.protobuf(tx)), Duration.Inf)
+      Await.result(grpcTxApi.broadcast(PBTransactions.protobuf(tx)), 10.seconds)
     }
   }
 }

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

@@ -420,7 +420,7 @@ class BlockchainUpdatesSpec extends FreeSpec with WithBUDomain with ScalaFutures
       (1 to blocksCount + 1).foreach(_ => d.appendBlock())
 
       val result = Await
-        .result(r.getBlockUpdatesRange(GetBlockUpdatesRangeRequest(1, blocksCount)), Duration.Inf)
+        .result(r.getBlockUpdatesRange(GetBlockUpdatesRangeRequest(1, blocksCount)), 1.minute)
         .updates
         .map(_.update.append.map(_.getBlock.vrf.toByteStr).filterNot(_.isEmpty))
 
@@ -1215,7 +1215,7 @@ class BlockchainUpdatesSpec extends FreeSpec with WithBUDomain with ScalaFutures
     Await
       .result(
         repo.getBlockUpdate(GetBlockUpdateRequest(height)),
-        Duration.Inf
+        1.minute
       )
       .getUpdate
       .update

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

@@ -13,6 +13,7 @@ import com.wavesplatform.transaction.Asset.IssuedAsset
 import com.wavesplatform.utils.ScorexLogging
 
 import java.nio.ByteBuffer
+import scala.collection.immutable.VectorMap
 import scala.collection.mutable.ArrayBuffer
 import scala.jdk.CollectionConverters.*
 
@@ -29,9 +30,8 @@ class NFTIterator(addressId: AddressId, maybeAfter: Option[IssuedAsset], resourc
       def skipEntry(key: Array[Byte]): Boolean =
         !key.endsWith(after.id.arr)
 
-      while (dbIterator.isValid && skipEntry(dbIterator.key())) {
+      while (dbIterator.isValid && skipEntry(dbIterator.key()))
         dbIterator.next()
-      }
       if (dbIterator.isValid && !skipEntry(dbIterator.key()))
         dbIterator.next()
     }
@@ -99,19 +99,14 @@ class WavesBalanceIterator(addressId: AddressId, resource: DBResource) extends A
 class BalanceIterator(
     address: Address,
     underlying: Iterator[Seq[(IssuedAsset, Long)]],
-    includeAsset: IssuedAsset => Boolean,
-    private var pendingOverrides: Map[(Address, Asset), Long]
+    private var pendingOverrides: VectorMap[(Address, Asset), Long]
 ) extends AbstractIterator[Seq[(IssuedAsset, Long)]] {
-
   private def nextOverride(): Seq[(IssuedAsset, Long)] =
     if (pendingOverrides.isEmpty) endOfData()
     else {
-      val balances = pendingOverrides.collect {
-        case ((`address`, asset: IssuedAsset), balance) if includeAsset(asset) =>
-          asset -> balance
-      }.toSeq
-      pendingOverrides = Map.empty
-      balances
+      val assetsWithBalances = pendingOverrides.collect { case ((`address`, asset: IssuedAsset), balance) => asset -> balance }.toSeq
+      pendingOverrides = VectorMap.empty
+      assetsWithBalances
     }
 
   override def computeNext(): Seq[(IssuedAsset, Long)] =
@@ -138,11 +133,15 @@ object AddressPortfolio {
       resource
         .get(Keys.addressId(address))
         .fold(Iterator.empty[Seq[(IssuedAsset, Long)]])(addressId => new NFTIterator(addressId, maybeAfter, resource).asScala),
-      asset => loadAssetDescription(asset).exists(_.nft),
       snapshot.balances
     ).asScala
-      .map(_.collect { case (asset, balance) if balance > 0 => asset }
-        .flatMap(a => loadAssetDescription(a).map(a -> _)))
+      .map { assets =>
+        maybeAfter
+          .filter(after => assets.exists(_._1 == after))
+          .fold(assets)(after => assets.dropWhile(_._1 != after).drop(1))
+          .collect { case (asset, balance) if balance > 0 => asset }
+          .flatMap(asset => loadAssetDescription(asset).collect { case description if description.nft => asset -> description })
+      }
 
   def assetBalanceIterator(
       resource: DBResource,
@@ -155,10 +154,9 @@ object AddressPortfolio {
       resource
         .get(Keys.addressId(address))
         .fold(Iterator.empty[Seq[(IssuedAsset, Long)]])(addressId => new AssetBalanceIterator(addressId, resource).asScala),
-      includeAsset,
       snapshot.balances
     ).asScala
-      .map(_.filter { case (asset, balance) =>
-        includeAsset(asset) && balance > 0
-      })
+      .map(
+        _.filter { case (asset, balance) => includeAsset(asset) && balance > 0 }
+      )
 }

node/src/test/scala/com/wavesplatform/mining/BlockWithMaxBaseTargetTest.scala

@@ -102,7 +102,7 @@ class BlockWithMaxBaseTargetTest extends FreeSpec with WithNewDBForEachTest with
           case _: SecurityException =>
             Task.unit
         }
-        Await.result(blockAppendTask.runToFuture(scheduler), Duration.Inf)
+        Await.result(blockAppendTask.runToFuture(scheduler), 1.minute)
 
         signal.tryAcquire(10, TimeUnit.SECONDS)
 

node/src/test/scala/com/wavesplatform/state/LightNodeTest.scala

@@ -138,7 +138,15 @@ class LightNodeTest extends PropSpec with WithDomain {
         )
 
         val appender =
-          ExtensionAppender(d.blockchain, d.utxPool, d.posSelector, TestTime(), InvalidBlockStorage.NoOp, PeerDatabase.NoOp, Scheduler.global)(
+          ExtensionAppender(
+            d.blockchain,
+            d.utxPool,
+            d.posSelector,
+            TestTime(extensionBlocks.blocks.last.header.timestamp),
+            InvalidBlockStorage.NoOp,
+            PeerDatabase.NoOp,
+            Scheduler.global
+          )(
             null,
             _
           )
@@ -162,7 +170,7 @@ class LightNodeTest extends PropSpec with WithDomain {
       val challengingBlock = d.createChallengingBlock(challengingMiner, invalidBlock, strictTime = true)
       val txSnapshots      = getTxSnapshots(d, challengingBlock)
 
-      val appender = BlockAppender(d.blockchainUpdater, TestTime(), d.utxPool, d.posSelector, Scheduler.global) _
+      val appender = BlockAppender(d.blockchainUpdater, TestTime(challengingBlock.header.timestamp), d.utxPool, d.posSelector, Scheduler.global) _
 
       appender(challengingBlock, Some(BlockSnapshot(challengingBlock.id(), txSnapshots))).runSyncUnsafe() shouldBe Right(
         Applied(Seq.empty, d.blockchain.score)

node/src/test/scala/com/wavesplatform/state/TransactionsByAddressSpec.scala

@@ -21,7 +21,6 @@ import java.util.concurrent.locks.ReentrantLock
 import scala.concurrent.ExecutionContext.Implicits.global
 import scala.concurrent.{Await, Future}
 import scala.concurrent.duration.*
-import scala.concurrent.duration.Duration.Inf
 
 class TransactionsByAddressSpec extends FreeSpec with BlockGen with WithDomain {
   def transfers(sender: KeyPair, rs: AddressOrAlias, amount: Long): Seq[TransferTransaction] =
@@ -122,7 +121,7 @@ class TransactionsByAddressSpec extends FreeSpec with BlockGen with WithDomain {
         val txs = Future { d.addressTransactions(defaultAddress).map(_._2.tpe) }
         d.blockchain.bestLiquidSnapshot.synchronized(d.appendKeyBlock())
         startRead.unlock()
-        Await.result(txs, Inf).map(_.tpe) shouldBe List(TransactionType.Issue)
+        Await.result(txs, 1.minute).map(_.tpe) shouldBe List(TransactionType.Issue)
       }
     }
   }

node/src/test/scala/com/wavesplatform/state/diffs/ExchangeTransactionDiffTest.scala

@@ -1928,7 +1928,7 @@ class ExchangeTransactionDiffTest extends PropSpec with Inside with WithDomain w
         TxExchangeAmount(1),
         TxOrderPrice(1),
         System.currentTimeMillis(),
-        System.currentTimeMillis() + 1.day.toMillis,
+        System.currentTimeMillis() + 10.hours.toMillis,
         TxMatcherFee.unsafeFrom(0.003.waves)
       )
       val signedBuyOrder = buyOrder.copy(
@@ -1994,7 +1994,7 @@ class ExchangeTransactionDiffTest extends PropSpec with Inside with WithDomain w
         TxExchangeAmount(1),
         TxOrderPrice(1),
         System.currentTimeMillis(),
-        System.currentTimeMillis() + 10000,
+        System.currentTimeMillis() + 10.hours.toMillis,
         TxMatcherFee.unsafeFrom(0.003.waves)
       )
       val signature = EthOrders.signOrder(buyOrder, signer)