ExtensionAppender refactoring

vsuharnikov · vsuharnikov · commit b837f9825b24 · 2026-04-03T19:01:39.000+04:00
diff --git a/node/src/main/scala/com/wavesplatform/state/ParSignatureChecker.scala b/node/src/main/scala/com/wavesplatform/state/ParSignatureChecker.scala
@@ -7,6 +7,10 @@ import com.wavesplatform.utils.Schedulers
 import monix.eval.Task
 import monix.execution.schedulers.SchedulerService
 
+/** Executes signature verification in advance using parallel computation.
+  * The result is cached in a lazy value: if precomputation completes in time,
+  * the cached result is reused; otherwise, it is computed on demand.
+  */
 object ParSignatureChecker {
   implicit val sigverify: SchedulerService = Schedulers.fixedPool(4, "sigverify")
 
diff --git a/node/src/main/scala/com/wavesplatform/state/appender/ExtensionAppender.scala b/node/src/main/scala/com/wavesplatform/state/appender/ExtensionAppender.scala
@@ -1,5 +1,10 @@
 package com.wavesplatform.state.appender
 
+import cats.data.OptionT
+import cats.syntax.all.*
+import com.wavesplatform.block.Block
+import com.wavesplatform.block.Block.BlockId
+import com.wavesplatform.common.state.ByteStr
 import com.wavesplatform.common.utils.EitherExt2.*
 import com.wavesplatform.consensus.PoSSelector
 import com.wavesplatform.features.BlockchainFeatures
@@ -17,9 +22,11 @@ import monix.eval.Task
 import monix.execution.Scheduler
 import org.influxdb.dto.Point
 
-import scala.util.{Left, Right}
+import scala.annotation.tailrec
+import scala.util.chaining.*
 
 object ExtensionAppender extends ScorexLogging {
+  private case class AppendData(newBlocks: Seq[Block], lastCommonBlockId: BlockId, lastCommonHeight: Int)
 
   def apply(
       blockchainUpdater: BlockchainUpdater & Blockchain,
@@ -29,111 +36,125 @@ object ExtensionAppender extends ScorexLogging {
       invalidBlocks: InvalidBlockStorage,
       peerDatabase: PeerDatabase,
       scheduler: Scheduler
-  )(ch: Channel, extensionBlocks: ExtensionBlocks): Task[Either[ValidationError, Option[BigInt]]] = {
-    def appendExtension(extension: ExtensionBlocks): Either[ValidationError, Option[BigInt]] =
+  )(ch: Channel, extension: ExtensionBlocks): Task[Either[ValidationError, Option[BigInt]]] = {
+    type Result[A] = Either[ValidationError, A]
+    def appendExtension(): OptionT[Result, BigInt] =
       if (extension.remoteScore <= blockchainUpdater.score) {
         log.trace(s"Ignoring extension $extension because declared remote was not greater than local score ${blockchainUpdater.score}")
-        Right(None)
-      } else {
-        extension.blocks
-          .collectFirst { case b if !b.signatureValid() => GenericError(s"Block $b has invalid signature") }
-          .toLeft(extension)
-          .flatMap { extensionWithValidSignatures =>
-            val newBlocks = extensionWithValidSignatures.blocks.dropWhile(blockchainUpdater.contains)
-
-            newBlocks.headOption.map(_.header.reference) match {
-              case Some(lastCommonBlockId) =>
-                val initialHeight = blockchainUpdater.height
-
-                val droppedBlocksEi = for {
-                  commonBlockHeight <- blockchainUpdater.heightOf(lastCommonBlockId).toRight(GenericError("Fork contains no common parent"))
-                  droppedBlocks <- {
-                    if (commonBlockHeight < initialHeight)
-                      blockchainUpdater.removeAfter(lastCommonBlockId)
-                    else Right(Seq.empty)
-                  }
-                } yield (commonBlockHeight, droppedBlocks)
-
-                droppedBlocksEi.flatMap { case (commonBlockHeight, droppedBlocks) =>
-                  newBlocks.zipWithIndex.foreach { case (block, idx) =>
-                    val rideV6Activated = blockchainUpdater.isFeatureActivated(BlockchainFeatures.RideV6, commonBlockHeight + idx + 1)
-                    ParSignatureChecker.checkTxSignatures(block.transactionData, rideV6Activated)
-                  }
-
-                  val forkApplicationResultEi = {
-                    newBlocks.view
-                      .map { b =>
-                        b -> appendExtensionBlock(blockchainUpdater, pos, time, verify = true, txSignParCheck = false)(
-                          b,
-                          extension.snapshots.get(b.id())
-                        )
-                          .map {
-                            case (_: Applied, height) => BlockStats.applied(b, BlockStats.Source.Ext, height)
-                            case _                    =>
-                          }
-                      }
-                      .zipWithIndex
-                      .collectFirst { case ((b, Left(e)), i) => (i, b, e) }
-                      .fold[Either[ValidationError, Unit]](Right(())) { case (i, declinedBlock, e) =>
-                        e match {
-                          case _: TxValidationError.BlockFromFuture =>
-                          case _                                    => invalidBlocks.add(declinedBlock.id(), e)
-                        }
-
-                        newBlocks.view
-                          .dropWhile(_ != declinedBlock)
-                          .foreach(BlockStats.declined(_, BlockStats.Source.Ext))
-
-                        if (i == 0) log.warn(s"Can't process fork starting with $lastCommonBlockId, error appending block $declinedBlock: $e")
-                        else
-                          log.warn(
-                            s"Processed only ${i + 1} of ${newBlocks.size} blocks from extension, error appending next block $declinedBlock: $e"
-                          )
-
-                        Left(e)
-                      }
-                  }
-
-                  forkApplicationResultEi match {
-                    case Left(e) =>
-                      blockchainUpdater.removeAfter(lastCommonBlockId).explicitGet()
-                      droppedBlocks.foreach { x =>
-                        blockchainUpdater.processBlock(x.block, x.hitSource, x.snapshot, x.generatorSet).explicitGet()
-                      }
-                      Left(e)
-
-                    case Right(_) =>
-                      val depth = initialHeight - commonBlockHeight
-                      if (depth > 0) {
-                        Metrics.write(
-                          Point
-                            .measurement("rollback")
-                            .addField("depth", initialHeight - commonBlockHeight)
-                            .addField("txs", droppedBlocks.size)
-                        )
-                      }
-
-                      val newTransactions = newBlocks.view.flatMap(_.transactionData).toSet
-                      utxStorage.removeAll(newTransactions)
-                      utxStorage.addAndScheduleCleanup(droppedBlocks.flatMap(_._1.transactionData).filterNot(newTransactions))
-                      Right(Some(blockchainUpdater.score))
-                  }
-                }
-
-              case None =>
-                log.debug("No new blocks found in extension")
-                Right(None)
-            }
+        OptionT.none[Result, BigInt]
+      } else
+        for {
+          _          <- OptionT.liftF(validateSignatures())
+          appendData <- OptionT.fromOption[Result](dropCommonPrefix())
+          _          <- OptionT.liftF(processNewBlocks(appendData))
+        } yield blockchainUpdater.score
+
+    def validateSignatures(): Either[ValidationError, ExtensionBlocks] =
+      extension.blocks
+        .collectFirst { case b if !b.signatureValid() => GenericError(s"Block $b has invalid signature") }
+        .toLeft(extension)
+
+    def dropCommonPrefix(): Option[AppendData] = {
+      @tailrec def loop(last: AppendData): Option[AppendData] = last.newBlocks match {
+        case Nil =>
+          log.debug("No new blocks found in extension")
+          None
+
+        case b +: rest =>
+          blockchainUpdater.heightOf(b.id()) match {
+            case None    => last.some
+            case Some(h) => loop(AppendData(rest, b.id(), h))
           }
       }
 
-    log.debug(s"${id(ch)} Attempting to append extension ${formatBlocks(extensionBlocks.blocks)}")
-    Task(appendExtension(extensionBlocks)).executeOn(scheduler).map {
+      loop(
+        AppendData(
+          extension.blocks,
+          blockchainUpdater.lastBlockId.getOrElse(throw new RuntimeException("Empty blockchain")),
+          blockchainUpdater.height
+        )
+      )
+    }
+
+    def processNewBlocks(appendData: AppendData): Either[ValidationError, Unit] = {
+      val originalForkHeight = blockchainUpdater.height
+      for {
+        discardedBlocks <-
+          if (appendData.lastCommonHeight < originalForkHeight) blockchainUpdater.removeAfter(appendData.lastCommonBlockId)
+          else Right(Seq.empty)
+        _ = precheckSignatures(appendData)
+        _ <- applyFork(appendData).tap {
+          case Left(_) => restoreDiscardedBlocks(appendData.lastCommonBlockId, discardedBlocks)
+          case Right(_) =>
+            val depth = originalForkHeight - appendData.lastCommonHeight
+            if (depth > 0)
+              Metrics.write(
+                Point
+                  .measurement("rollback")
+                  .addField("depth", depth)
+                  .addField("txs", discardedBlocks.size)
+              )
+
+            val newTxs   = appendData.newBlocks.flatMap(_.transactionData)
+            val newTxIds = newTxs.view.map(_.id()).toSet
+            utxStorage.removeIds(newTxIds)
+
+            val discardedTxs = discardedBlocks.flatMap(_._1.transactionData)
+            utxStorage.addAndScheduleCleanup(discardedTxs.filterNot(tx => newTxIds.contains(tx.id()))) // In a case of re-appending issues
+        }
+      } yield ()
+    }
+
+    def precheckSignatures(appendData: AppendData): Unit =
+      appendData.newBlocks.zipWithIndex.foreach { case (block, idx) =>
+        val rideV6Activated = blockchainUpdater.isFeatureActivated(BlockchainFeatures.RideV6, appendData.lastCommonHeight + idx + 1)
+        ParSignatureChecker.checkTxSignatures(block.transactionData, rideV6Activated)
+      }
+
+    def applyFork(appendData: AppendData): Either[ValidationError, Unit] = appendData.newBlocks.view
+      .map { b =>
+        val s = extension.snapshots.get(b.id())
+        val r = appendExtensionBlock(blockchainUpdater, pos, time, verify = true, txSignParCheck = false)(b, s).map {
+          case (_: Applied, height) => BlockStats.applied(b, BlockStats.Source.Ext, height)
+          case _                    =>
+        }
+        b -> r
+      }
+      .zipWithIndex
+      .collectFirst { case ((b, Left(e)), i) => (i, b, e) }
+      .fold(Either.unit[ValidationError]) { case (i, declinedBlock, e) =>
+        e match {
+          case _: TxValidationError.BlockFromFuture =>
+          case _                                    => invalidBlocks.add(declinedBlock.id(), e)
+        }
+
+        appendData.newBlocks.view
+          .dropWhile(_ != declinedBlock)
+          .foreach(BlockStats.declined(_, BlockStats.Source.Ext))
+
+        log.warn(
+          if (i == 0) s"Can't process fork starting with ${appendData.lastCommonBlockId}, error appending block $declinedBlock: $e"
+          else s"Processed only ${i + 1} of ${appendData.newBlocks.size} blocks from extension, error appending next block $declinedBlock: $e"
+        )
+
+        Left(e)
+      }
+
+    def restoreDiscardedBlocks(lastCommonBlockId: ByteStr, blocks: DiscardedBlocks): Unit = {
+      blockchainUpdater.removeAfter(lastCommonBlockId).explicitGet()
+      blocks.foreach { x =>
+        blockchainUpdater.processBlock(x.block, x.hitSource, x.snapshot, x.generatorSet).explicitGet()
+      }
+    }
+
+    val formattedBlocksStr = formatBlocks(extension.blocks)
+    log.debug(s"${id(ch)} Attempting to append extension $formattedBlocksStr")
+    Task(appendExtension().value).executeOn(scheduler).map {
       case Right(maybeNewScore) =>
-        log.debug(s"${id(ch)} Successfully appended extension ${formatBlocks(extensionBlocks.blocks)}")
+        log.debug(s"${id(ch)} Successfully appended extension $formattedBlocksStr")
         Right(maybeNewScore)
       case Left(ve) =>
-        val errorMessage = s"${id(ch)} Error appending extension ${formatBlocks(extensionBlocks.blocks)}: $ve"
+        val errorMessage = s"${id(ch)} Error appending extension $formattedBlocksStr: $ve"
         log.warn(errorMessage)
         peerDatabase.blacklistAndClose(ch, errorMessage)
         Left(ve)
diff --git a/node/src/main/scala/com/wavesplatform/utx/UtxPool.scala b/node/src/main/scala/com/wavesplatform/utx/UtxPool.scala
@@ -16,7 +16,7 @@ trait UtxForAppender {
 
 trait UtxPool extends UtxForAppender with AutoCloseable {
   def putIfNew(tx: Transaction, forceValidate: Boolean = false): TracedResult[ValidationError, Boolean]
-  def removeAll(txs: Iterable[Transaction]): Unit
+  def removeIds(txIds: Iterable[ByteStr]): Unit
   def all: Seq[Transaction]
   def size: Int
   def transactionById(transactionId: ByteStr): Option[Transaction]
@@ -40,4 +40,11 @@ object UtxPool {
     case class Estimate(time: FiniteDuration) extends PackStrategy
     case object Unlimited                     extends PackStrategy
   }
+
+  extension (self: UtxPool) {
+    def removeAll(txs: Iterable[Transaction]): Unit = if (txs.nonEmpty) {
+      val ids = txs.map(_.id()).toSet
+      self.removeIds(ids)
+    }
+  }
 }
diff --git a/node/src/main/scala/com/wavesplatform/utx/UtxPoolImpl.scala b/node/src/main/scala/com/wavesplatform/utx/UtxPoolImpl.scala
@@ -16,8 +16,7 @@ import com.wavesplatform.state.TxStateSnapshotHashBuilder.TxStatusInfo
 import com.wavesplatform.state.diffs.BlockDiffer.CurrentBlockFeePart
 import com.wavesplatform.state.diffs.TransactionDiffer.TransactionValidationError
 import com.wavesplatform.state.diffs.{BlockDiffer, TransactionDiffer}
-import com.wavesplatform.state.SnapshotBlockchain
-import com.wavesplatform.state.{Blockchain, Portfolio, StateSnapshot, TxStateSnapshotHashBuilder}
+import com.wavesplatform.state.{Blockchain, Portfolio, SnapshotBlockchain, StateSnapshot, TxStateSnapshotHashBuilder}
 import com.wavesplatform.transaction.*
 import com.wavesplatform.transaction.TxValidationError.{AlreadyInTheState, GenericError, SenderIsBlacklisted, WithLog}
 import com.wavesplatform.transaction.assets.exchange.ExchangeTransaction
@@ -166,12 +165,6 @@ case class UtxPoolImpl(
     tracedIsNew
   }
 
-  override def removeAll(txs: Iterable[Transaction]): Unit = {
-    if (txs.isEmpty) return
-    val ids = txs.map(_.id()).toSet
-    removeIds(ids)
-  }
-
   def setPrioritySnapshots(discSnapshots: Seq[StateSnapshot]): Unit =
     priorityPool.setPriorityDiffs(discSnapshots).foreach(addTransaction(_, verify = false))
 
@@ -185,8 +178,8 @@ case class UtxPoolImpl(
     }
   }
 
-  private def removeIds(removed: Set[ByteStr]): Unit =
-    removed.flatMap(id => removeFromOrdPool(id)).foreach(TxStateActions.removeMined(_))
+  def removeIds(txIds: Iterable[ByteStr]): Unit =
+    txIds.flatMap(id => removeFromOrdPool(id)).foreach(TxStateActions.removeMined(_))
 
   private[utx] def addTransaction(
       tx: Transaction,
diff --git a/node/tests/src/test/scala/com/wavesplatform/finalization/conflict/ConflictEndorserBlocksNgSuite.scala b/node/tests/src/test/scala/com/wavesplatform/finalization/conflict/ConflictEndorserBlocksNgSuite.scala
@@ -14,7 +14,7 @@ import com.wavesplatform.transaction.TxHelpers
 import org.scalactic.source.Position
 import org.scalatest.Assertion
 
-/** Blocks:
+/** Adding blocks:
   * 1. Genesis
   * 2. With commitments from two generators
   * 3. First block at period #1
diff --git a/node/tests/src/test/scala/com/wavesplatform/mining/MicroBlockMinerSpec.scala b/node/tests/src/test/scala/com/wavesplatform/mining/MicroBlockMinerSpec.scala
@@ -123,7 +123,6 @@ class MicroBlockMinerSpec extends FlatSpec with WithDomain {
       )
 
       val utxPool = new UtxPool {
-
         override def packUnconfirmed(
             rest: MultiDimensionalMiningConstraint,
             prevStateHash: Option[ByteStr],
@@ -140,7 +139,7 @@ class MicroBlockMinerSpec extends FlatSpec with WithDomain {
         }
 
         override def putIfNew(tx: Transaction, forceValidate: Boolean)                = inner.putIfNew(tx, forceValidate)
-        override def removeAll(txs: Iterable[Transaction]): Unit                      = inner.removeAll(txs)
+        override def removeIds(txIds: Iterable[ByteStr]): Unit                        = inner.removeIds(txIds)
         override def all                                                              = inner.all
         override def size                                                             = inner.size
         override def transactionById(transactionId: ByteStr)                          = inner.transactionById(transactionId)
diff --git a/node/tests/src/test/scala/com/wavesplatform/state/LightNodeTest.scala b/node/tests/src/test/scala/com/wavesplatform/state/LightNodeTest.scala
@@ -114,8 +114,9 @@ class LightNodeTest extends PropSpec with WithDomain {
         DomainPresets.TransactionStateSnapshot.copy(enableLightMode = isLightMode),
         AddrWithBalance.enoughBalances(sender, TxHelpers.defaultSigner)
       ) { d =>
-        val chainSize = 3
-        val genesisId = d.lastBlockId
+        val chainSize    = 3
+        val genesisId    = d.lastBlockId
+        val genesisBlock = d.lastBlock
         val betterBlocks = (1 to chainSize).map { idx =>
           val txs =
             Seq(TxHelpers.transfer(sender, recipient, amount = (idx + 10).waves), TxHelpers.transfer(sender, recipient, amount = (idx + 11).waves))
@@ -125,17 +126,21 @@ class LightNodeTest extends PropSpec with WithDomain {
           block -> txSnapshots
         }
         val expectedStateHash = d.lastBlock.header.stateHash
+
+        log.debug("Rolling back")
         d.rollbackTo(genesisId)
 
+        log.debug("Appending new blocks")
         (1 to chainSize).foreach { idx =>
           val txs = Seq(TxHelpers.transfer(sender, recipient, amount = idx.waves), TxHelpers.transfer(sender, recipient, (idx + 1).waves))
           d.appendBlock(txs*)
         }
         val currentScore = d.blockchain.score
 
+        log.debug("Appending extension")
         val extensionBlocks = ExtensionBlocks(
           currentScore + 1,
-          betterBlocks.map(_._1),
+          genesisBlock +: betterBlocks.map(_._1),
           betterBlocks.collect { case (b, Some(snapshots)) =>
             b.id() -> BlockSnapshotResponse(b.id(), snapshots.map { case (s, m) => PBSnapshots.toProtobuf(s, m) })
           }.toMap
diff --git a/node/tests/src/test/scala/com/wavesplatform/state/appender/ExtensionAppenderSpec.scala b/node/tests/src/test/scala/com/wavesplatform/state/appender/ExtensionAppenderSpec.scala
@@ -14,6 +14,7 @@ import monix.execution.Scheduler.Implicits.global
 class ExtensionAppenderSpec extends FlatSpec with WithDomain {
   "Extension appender" should "drop duplicate transactions from UTX" in
     withDomain(balances = AddrWithBalance.enoughBalances(TxHelpers.defaultSigner)) { d =>
+      val genesisBlock = d.lastBlock
       val utx  = new UtxPoolImpl(SystemTime, d.blockchain, d.settings.utxSettings, d.settings.maxTxErrorLogSize, d.settings.minerSettings.enable)
       val time = TestTime()
       val extensionAppender = ExtensionAppender(d.blockchain, utx, d.posSelector, time, InvalidBlockStorage.NoOp, PeerDatabase.NoOp, global)(null, _)
@@ -25,7 +26,7 @@ class ExtensionAppenderSpec extends FlatSpec with WithDomain {
       utx.all shouldBe Seq(tx)
 
       time.setTime(block1.header.timestamp)
-      extensionAppender(ExtensionBlocks(d.blockchain.score + block1.blockScore(), Seq(block1), Map.empty)).runSyncUnsafe().explicitGet()
+      extensionAppender(ExtensionBlocks(d.blockchain.score + block1.blockScore(), Seq(genesisBlock, block1), Map.empty)).runSyncUnsafe().explicitGet()
       d.blockchain.height shouldBe 2
       utx.all shouldBe Nil
       utx.close()

Original file line number	Diff line number	Diff line change
`@@ -123,7 +123,6 @@ class MicroBlockMinerSpec extends FlatSpec with WithDomain {`
`123`	`123`	`)`
`124`	`124`
`125`	`125`	`val utxPool = new UtxPool {`
`126`		`-`
`127`	`126`	`override def packUnconfirmed(`
`128`	`127`	`rest: MultiDimensionalMiningConstraint,`
`129`	`128`	`prevStateHash: Option[ByteStr],`
`@@ -140,7 +139,7 @@ class MicroBlockMinerSpec extends FlatSpec with WithDomain {`
`140`	`139`	`}`
`141`	`140`
`142`	`141`	`override def putIfNew(tx: Transaction, forceValidate: Boolean) = inner.putIfNew(tx, forceValidate)`
`143`		`- override def removeAll(txs: Iterable[Transaction]): Unit = inner.removeAll(txs)`
	`142`	`+ override def removeIds(txIds: Iterable[ByteStr]): Unit = inner.removeIds(txIds)`
`144`	`143`	`override def all = inner.all`
`145`	`144`	`override def size = inner.size`
`146`	`145`	`override def transactionById(transactionId: ByteStr) = inner.transactionById(transactionId)`