NODE-2348 Consistent API liquid state

node/src/main/scala/com/wavesplatform/Application.scala

@@ -17,6 +17,7 @@ import com.wavesplatform.api.http.assets.AssetsApiRoute
 import com.wavesplatform.api.http.eth.EthRpcRoute
 import com.wavesplatform.api.http.leasing.LeaseApiRoute
 import com.wavesplatform.common.state.ByteStr
+import com.wavesplatform.common.utils.EitherExt2
 import com.wavesplatform.consensus.PoSSelector
 import com.wavesplatform.database.{DBExt, Keys, openDB}
 import com.wavesplatform.events.{BlockchainUpdateTriggers, UtxEvent}
@@ -30,6 +31,7 @@ import com.wavesplatform.mining.{Miner, MinerDebugInfo, MinerImpl}
 import com.wavesplatform.network.*
 import com.wavesplatform.settings.WavesSettings
 import com.wavesplatform.state.appender.{BlockAppender, ExtensionAppender, MicroblockAppender}
+import com.wavesplatform.state.reader.CompositeBlockchain
 import com.wavesplatform.state.{Blockchain, BlockchainUpdaterImpl, Diff, Height, TxMeta}
 import com.wavesplatform.transaction.TxValidationError.GenericError
 import com.wavesplatform.transaction.smart.script.trace.TracedResult
@@ -122,10 +124,20 @@ class Application(val actorSystem: ActorSystem, val settings: WavesSettings, con
 
     val establishedConnections = new ConcurrentHashMap[Channel, PeerInfo]
     val allChannels            = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE)
-    val utxStorage =
-      new UtxPoolImpl(time, blockchainUpdater, settings.utxSettings, settings.maxTxErrorLogSize, settings.minerSettings.enable, utxEvents.onNext)
+    val utxStorage             = new UtxPoolImpl(time, blockchainUpdater, settings.utxSettings, settings.maxTxErrorLogSize, settings.minerSettings.enable, utxEvents.onNext)
     maybeUtx = Some(utxStorage)
 
+    def blockchainWithDiscardedDiffs(): CompositeBlockchain = {
+      def blockchain = CompositeBlockchain(blockchainUpdater, utxStorage.discardedMicrosDiff())
+      utxStorage.priorityPool.optimisticRead(blockchain)(_ => true)
+    }
+
+    def totalLiquidDiff(): Diff = {
+      def liquidDiff = blockchainUpdater.bestLiquidDiff.getOrElse(Diff())
+      def totalDiff  = liquidDiff.combineE(utxStorage.discardedMicrosDiff()).explicitGet()
+      utxStorage.priorityPool.optimisticRead(totalDiff)(_ => true)
+    }
+
     val timer                 = new HashedWheelTimer()
     val utxSynchronizerLogger = LoggerFacade(LoggerFactory.getLogger(classOf[TransactionPublisher]))
     val timedTxValidator =
@@ -222,18 +234,17 @@ class Application(val actorSystem: ActorSystem, val settings: WavesSettings, con
       override def utxEvents: Observable[UtxEvent]                       = app.utxEvents
 
       override val transactionsApi: CommonTransactionsApi = CommonTransactionsApi(
-        blockchainUpdater.bestLiquidDiff.map(diff => Height(blockchainUpdater.height) -> diff),
+        Some(Height(blockchainUpdater.height) -> totalLiquidDiff()),
         db,
-        blockchainUpdater,
+        () => blockchainWithDiscardedDiffs(),
         utxStorage,
         tx => transactionPublisher.validateAndBroadcast(tx, None),
         loadBlockAt(db, blockchainUpdater)
       )
       override val blocksApi: CommonBlocksApi =
         CommonBlocksApi(blockchainUpdater, loadBlockMetaAt(db, blockchainUpdater), loadBlockInfoAt(db, blockchainUpdater))
-      override val accountsApi: CommonAccountsApi =
-        CommonAccountsApi(() => blockchainUpdater.bestLiquidDiff.getOrElse(Diff.empty), db, blockchainUpdater)
-      override val assetsApi: CommonAssetsApi = CommonAssetsApi(() => blockchainUpdater.bestLiquidDiff.getOrElse(Diff.empty), db, blockchainUpdater)
+      override val accountsApi: CommonAccountsApi = CommonAccountsApi(() => totalLiquidDiff(), db, () => blockchainWithDiscardedDiffs())
+      override val assetsApi: CommonAssetsApi     = CommonAssetsApi(() => totalLiquidDiff(), db, () => blockchainWithDiscardedDiffs())
     }
 
     extensions = settings.extensions.map { extensionClassName =>
@@ -360,14 +371,14 @@ class Application(val actorSystem: ActorSystem, val settings: WavesSettings, con
       )(heavyRequestScheduler)
 
       val apiRoutes = Seq(
-        new EthRpcRoute(blockchainUpdater, extensionContext.transactionsApi, time),
+        EthRpcRoute(() => blockchainWithDiscardedDiffs(), extensionContext.transactionsApi, time),
         NodeApiRoute(settings.restAPISettings, blockchainUpdater, () => shutdown()),
         BlocksApiRoute(settings.restAPISettings, extensionContext.blocksApi, time, routeTimeout),
         TransactionsApiRoute(
           settings.restAPISettings,
           extensionContext.transactionsApi,
           wallet,
-          blockchainUpdater,
+          () => blockchainWithDiscardedDiffs(),
           () => utxStorage.size,
           transactionPublisher,
           time,
@@ -386,7 +397,7 @@ class Application(val actorSystem: ActorSystem, val settings: WavesSettings, con
         AddressApiRoute(
           settings.restAPISettings,
           wallet,
-          blockchainUpdater,
+          () => blockchainWithDiscardedDiffs(),
           transactionPublisher,
           time,
           limitedScheduler,
@@ -422,7 +433,7 @@ class Application(val actorSystem: ActorSystem, val settings: WavesSettings, con
           settings.restAPISettings,
           wallet,
           transactionPublisher,
-          blockchainUpdater,
+          () => blockchainWithDiscardedDiffs(),
           time,
           extensionContext.accountsApi,
           extensionContext.assetsApi,
@@ -433,7 +444,6 @@ class Application(val actorSystem: ActorSystem, val settings: WavesSettings, con
         LeaseApiRoute(
           settings.restAPISettings,
           wallet,
-          blockchainUpdater,
           transactionPublisher,
           time,
           extensionContext.accountsApi,
@@ -445,7 +455,7 @@ class Application(val actorSystem: ActorSystem, val settings: WavesSettings, con
           wallet,
           transactionPublisher,
           time,
-          blockchainUpdater,
+          () => blockchainWithDiscardedDiffs(),
           routeTimeout
         ),
         RewardApiRoute(blockchainUpdater)

node/src/main/scala/com/wavesplatform/Importer.scala

@@ -134,17 +134,17 @@ object Importer extends ScorexLogging {
             CommonTransactionsApi(
               blockchainUpdater.bestLiquidDiff.map(diff => Height(blockchainUpdater.height) -> diff),
               db,
-              blockchainUpdater,
+              () => blockchainUpdater,
               utxPool,
               _ => Future.successful(TracedResult.wrapE(Left(GenericError("Not implemented during import")))),
               Application.loadBlockAt(db, blockchainUpdater)
             )
           override def blocksApi: CommonBlocksApi =
             CommonBlocksApi(blockchainUpdater, Application.loadBlockMetaAt(db, blockchainUpdater), Application.loadBlockInfoAt(db, blockchainUpdater))
           override def accountsApi: CommonAccountsApi =
-            CommonAccountsApi(() => blockchainUpdater.bestLiquidDiff.getOrElse(Diff.empty), db, blockchainUpdater)
+            CommonAccountsApi(() => blockchainUpdater.bestLiquidDiff.getOrElse(Diff.empty), db, () => blockchainUpdater)
           override def assetsApi: CommonAssetsApi =
-            CommonAssetsApi(() => blockchainUpdater.bestLiquidDiff.getOrElse(Diff.empty), db, blockchainUpdater)
+            CommonAssetsApi(() => blockchainUpdater.bestLiquidDiff.getOrElse(Diff.empty), db, () => blockchainUpdater)
         }
       }
 

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

@@ -55,21 +55,24 @@ object CommonAccountsApi {
 
   final case class BalanceDetails(regular: Long, generating: Long, available: Long, effective: Long, leaseIn: Long, leaseOut: Long)
 
-  def apply(diff: () => Diff, db: DB, blockchain: Blockchain): CommonAccountsApi = new CommonAccountsApi {
+  def apply(diff: () => Diff, db: DB, blockchain: () => Blockchain): CommonAccountsApi = new CommonAccountsApi {
 
-    override def balance(address: Address, confirmations: Int = 0): Long =
-      blockchain.balance(address, blockchain.height, confirmations)
+    override def balance(address: Address, confirmations: Int = 0): Long = {
+      val bc = blockchain()
+      bc.balance(address, bc.height, confirmations)
+    }
 
     override def effectiveBalance(address: Address, confirmations: Int = 0): Long = {
-      blockchain.effectiveBalance(address, confirmations)
+      blockchain().effectiveBalance(address, confirmations)
     }
 
     override def balanceDetails(address: Address): Either[String, BalanceDetails] = {
-      val portfolio = blockchain.wavesPortfolio(address)
+      val bc        = blockchain()
+      val portfolio = bc.wavesPortfolio(address)
       portfolio.effectiveBalance.map(effectiveBalance =>
         BalanceDetails(
           portfolio.balance,
-          blockchain.generatingBalance(address),
+          bc.generatingBalance(address),
           portfolio.balance - portfolio.lease.out,
           effectiveBalance,
           portfolio.lease.in,
@@ -78,26 +81,29 @@ object CommonAccountsApi {
       )
     }
 
-    override def assetBalance(address: Address, asset: IssuedAsset): Long = blockchain.balance(address, asset)
+    override def assetBalance(address: Address, asset: IssuedAsset): Long = blockchain().balance(address, asset)
 
     override def portfolio(address: Address): Observable[(IssuedAsset, Long)] = {
       val currentDiff = diff()
+      val bc          = blockchain()
       db.resourceObservable.flatMap { resource =>
-        Observable.fromIterator(Task(assetBalanceIterator(resource, address, currentDiff, includeNft(blockchain))))
+        Observable.fromIterator(Task(assetBalanceIterator(resource, address, currentDiff, includeNft(bc))))
       }
     }
 
     override def nftList(address: Address, after: Option[IssuedAsset]): Observable[(IssuedAsset, AssetDescription)] = {
       val currentDiff = diff()
+      val bc          = blockchain()
       db.resourceObservable.flatMap { resource =>
-        Observable.fromIterator(Task(nftIterator(resource, address, currentDiff, after, blockchain.assetDescription)))
+        Observable.fromIterator(Task(nftIterator(resource, address, currentDiff, after, bc.assetDescription)))
       }
     }
 
-    override def script(address: Address): Option[AccountScriptInfo] = blockchain.accountScript(address)
+    override def script(address: Address): Option[AccountScriptInfo] =
+      blockchain().accountScript(address)
 
     override def data(address: Address, key: String): Option[DataEntry[?]] =
-      blockchain.accountData(address, key)
+      blockchain().accountData(address, key)
 
     override def dataStream(address: Address, regex: Option[String]): Observable[DataEntry[?]] = Observable.defer {
       val pattern = regex.map(_.r.pattern)
@@ -124,12 +130,14 @@ object CommonAccountsApi {
       Observable.fromIterable((entriesFromDiff.values ++ entries).filterNot(_.isEmpty))
     }
 
-    override def resolveAlias(alias: Alias): Either[ValidationError, Address] = blockchain.resolveAlias(alias)
+    override def resolveAlias(alias: Alias): Either[ValidationError, Address] =
+      blockchain().resolveAlias(alias)
 
-    override def activeLeases(address: Address): Observable[LeaseInfo] =
+    override def activeLeases(address: Address): Observable[LeaseInfo] = {
+      val bc = blockchain()
       addressTransactions(
         db,
-        Some(Height(blockchain.height) -> diff()),
+        Some(Height(bc.height) -> diff()),
         address,
         None,
         Set(TransactionType.Lease, TransactionType.InvokeScript, TransactionType.InvokeExpression, TransactionType.Ethereum),
@@ -141,7 +149,7 @@ object CommonAccountsApi {
               lt.id(),
               lt.id(),
               lt.sender.toAddress,
-              blockchain.resolveAlias(lt.recipient).explicitGet(),
+              bc.resolveAlias(lt.recipient).explicitGet(),
               lt.amount.value,
               leaseHeight,
               LeaseInfo.Status.Active
@@ -153,13 +161,15 @@ object CommonAccountsApi {
           extractLeases(address, scriptResult, tx.id(), height)
         case _ => Seq()
       }
+    }
 
     private def extractLeases(subject: Address, result: InvokeScriptResult, txId: ByteStr, height: Height): Seq[LeaseInfo] = {
+      val bc = blockchain()
       (for {
         lease   <- result.leases
-        details <- blockchain.leaseDetails(lease.id) if details.isActive
+        details <- bc.leaseDetails(lease.id) if details.isActive
         sender = details.sender.toAddress
-        recipient <- blockchain.resolveAlias(lease.recipient).toOption if subject == sender || subject == recipient
+        recipient <- bc.resolveAlias(lease.recipient).toOption if subject == sender || subject == recipient
       } yield LeaseInfo(
         lease.id,
         txId,
@@ -180,26 +190,29 @@ object CommonAccountsApi {
           Right(recipientAddress)
         }
 
-    def leaseInfo(leaseId: ByteStr): Option[LeaseInfo] = blockchain.leaseDetails(leaseId) map { ld =>
-      LeaseInfo(
-        leaseId,
-        ld.sourceId,
-        ld.sender.toAddress,
-        blockchain.resolveAlias(ld.recipient).orElse(resolveDisabledAlias(leaseId)).explicitGet(),
-        ld.amount,
-        ld.height,
-        ld.status match {
-          case Status.Active          => LeaseInfo.Status.Active
-          case Status.Cancelled(_, _) => LeaseInfo.Status.Canceled
-          case Status.Expired(_)      => LeaseInfo.Status.Expired
-        },
-        ld.status.cancelHeight,
-        ld.status.cancelTransactionId
-      )
+    def leaseInfo(leaseId: ByteStr): Option[LeaseInfo] = {
+      val bc = blockchain()
+      bc.leaseDetails(leaseId) map { ld =>
+        LeaseInfo(
+          leaseId,
+          ld.sourceId,
+          ld.sender.toAddress,
+          bc.resolveAlias(ld.recipient).orElse(resolveDisabledAlias(leaseId)).explicitGet(),
+          ld.amount,
+          ld.height,
+          ld.status match {
+            case Status.Active          => LeaseInfo.Status.Active
+            case Status.Cancelled(_, _) => LeaseInfo.Status.Canceled
+            case Status.Expired(_)      => LeaseInfo.Status.Expired
+          },
+          ld.status.cancelHeight,
+          ld.status.cancelTransactionId
+        )
+      }
     }
 
     private[this] def leaseIsActive(id: ByteStr): Boolean =
-      blockchain.leaseDetails(id).exists(_.isActive)
+      blockchain().leaseDetails(id).exists(_.isActive)
   }
 
 }

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

@@ -23,26 +23,28 @@ trait CommonAssetsApi {
 object CommonAssetsApi {
   final case class AssetInfo(description: AssetDescription, issueTransaction: Option[IssueTransaction], sponsorBalance: Option[Long])
 
-  def apply(diff: () => Diff, db: DB, blockchain: Blockchain): CommonAssetsApi = new CommonAssetsApi {
+  def apply(diff: () => Diff, db: DB, blockchain: () => Blockchain): CommonAssetsApi = new CommonAssetsApi {
     def description(assetId: IssuedAsset): Option[AssetDescription] =
-      blockchain.assetDescription(assetId)
+      blockchain().assetDescription(assetId)
 
-    def fullInfo(assetId: IssuedAsset): Option[AssetInfo] =
+    def fullInfo(assetId: IssuedAsset): Option[AssetInfo] = {
+      val bc = blockchain()
       for {
-        assetInfo <- blockchain.assetDescription(assetId)
-        sponsorBalance = if (assetInfo.sponsorship != 0) Some(blockchain.wavesPortfolio(assetInfo.issuer.toAddress).spendableBalance) else None
+        assetInfo <- bc.assetDescription(assetId)
+        sponsorBalance = if (assetInfo.sponsorship != 0) Some(bc.wavesPortfolio(assetInfo.issuer.toAddress).spendableBalance) else None
       } yield AssetInfo(
         assetInfo,
-        blockchain.transactionInfo(assetId.id).collect { case (tm, it: IssueTransaction) if tm.succeeded => it },
+        bc.transactionInfo(assetId.id).collect { case (tm, it: IssueTransaction) if tm.succeeded => it },
         sponsorBalance
       )
+    }
 
     override def wavesDistribution(height: Int, after: Option[Address]): Observable[(Address, Long)] =
       balanceDistribution(
         db,
         height,
         after,
-        if (height == blockchain.height) diff().portfolios else Map.empty[Address, Portfolio],
+        if (height == blockchain().height) diff().portfolios else Map.empty[Address, Portfolio],
         KeyTags.WavesBalance.prefixBytes,
         bs => AddressId.fromByteArray(bs.slice(2, bs.length - 4)),
         _.balance
@@ -53,7 +55,7 @@ object CommonAssetsApi {
         db,
         height,
         after,
-        if (height == blockchain.height) diff().portfolios else Map.empty[Address, Portfolio],
+        if (height == blockchain().height) diff().portfolios else Map.empty[Address, Portfolio],
         KeyTags.AssetBalance.prefixBytes ++ asset.id.arr,
         bs => AddressId.fromByteArray(bs.slice(2 + crypto.DigestLength, bs.length - 4)),
         _.assets.getOrElse(asset, 0L)

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

@@ -47,7 +47,7 @@ object CommonTransactionsApi {
   def apply(
       maybeDiff: => Option[(Height, Diff)],
       db: DB,
-      blockchain: Blockchain,
+      blockchain: () => Blockchain,
       utx: UtxPool,
       publishTransaction: Transaction => Future[TracedResult[ValidationError, Boolean]],
       blockAt: Int => Option[(BlockMeta, Seq[(TxMeta, Transaction)])]
@@ -63,7 +63,7 @@ object CommonTransactionsApi {
       common.addressTransactions(db, maybeDiff, subject, sender, transactionTypes, fromId)
 
     override def transactionById(transactionId: ByteStr): Option[TransactionMeta] =
-      blockchain.transactionInfo(transactionId).map(common.loadTransactionMeta(db, maybeDiff))
+      blockchain().transactionInfo(transactionId).map(common.loadTransactionMeta(db, maybeDiff))
 
     override def unconfirmedTransactions: Seq[Transaction] = utx.all
 
@@ -72,7 +72,7 @@ object CommonTransactionsApi {
 
     override def calculateFee(tx: Transaction): Either[ValidationError, (Asset, Long, Long)] =
       FeeValidation
-        .getMinFee(blockchain, tx)
+        .getMinFee(blockchain(), tx)
         .map {
           case FeeDetails(asset, _, feeInAsset, feeInWaves) =>
             (asset, feeInAsset, feeInWaves)
@@ -83,7 +83,7 @@ object CommonTransactionsApi {
     override def transactionProofs(transactionIds: List[ByteStr]): List[TransactionProof] =
       for {
         transactionId           <- transactionIds
-        (txm, tx)               <- blockchain.transactionInfo(transactionId)
+        (txm, tx)               <- blockchain().transactionInfo(transactionId)
         (meta, allTransactions) <- blockAt(txm.height) if meta.header.version >= Block.ProtoBlockVersion
         transactionProof        <- block.transactionProof(tx, allTransactions.map(_._2))
       } yield transactionProof