P2P networking support for bootstrapping with UTXO set snapshot

avldb/src/main/scala/scorex/core/serialization/ManifestSerializer.scala

@@ -1,6 +1,5 @@
 package scorex.core.serialization
 
-import com.google.common.primitives.Ints
 import scorex.crypto.authds.avltree.batch.Constants.DigestType
 import scorex.crypto.authds.avltree.batch.serialization.{BatchAVLProverManifest, ProxyInternalNode}
 import scorex.crypto.authds.avltree.batch.{InternalProverNode, ProverLeaf, ProverNodes, VersionedLDBAVLStorage}
@@ -13,8 +12,8 @@ class ManifestSerializer(manifestDepth: Byte) extends ErgoSerializer[BatchAVLPro
   private val nodeSerializer = VersionedLDBAVLStorage.noStoreSerializer
 
   override def serialize(manifest: BatchAVLProverManifest[DigestType], w: Writer): Unit = {
-    val height = manifest.rootHeight
-    w.putBytes(Ints.toByteArray(height))
+    val rootNodeHeight = manifest.rootHeight.toByte
+    w.put(rootNodeHeight)
     w.put(manifestDepth)
 
     def loop(node: ProverNodes[DigestType], level: Int): Unit = {
@@ -32,10 +31,11 @@ class ManifestSerializer(manifestDepth: Byte) extends ErgoSerializer[BatchAVLPro
   }
 
   override def parse(r: Reader): BatchAVLProverManifest[DigestType] = {
-    val rootHeight = Ints.fromByteArray(r.getBytes(4))
+    val rootHeight = r.getByte()
     val manifestDepth = r.getByte()
 
-    require(manifestDepth == this.manifestDepth, "Wrong manifest depth")
+    require(manifestDepth == this.manifestDepth,
+            s"Wrong manifest depth, found: $manifestDepth, expected: ${this.manifestDepth}")
 
     def loop(level: Int): ProverNodes[DigestType] = {
       val node = nodeSerializer.parse(r)

avldb/src/main/scala/scorex/crypto/authds/avltree/batch/VersionedLDBAVLStorage.scala

@@ -1,7 +1,6 @@
 package scorex.crypto.authds.avltree.batch
 
 import com.google.common.primitives.Ints
-import scorex.core.serialization.ManifestSerializer
 import scorex.crypto.authds.avltree.batch.Constants.{DigestType, HashFnType, hashFn}
 import scorex.crypto.authds.avltree.batch.VersionedLDBAVLStorage.{topNodeHashKey, topNodeHeightKey}
 import scorex.crypto.authds.avltree.batch.serialization.{BatchAVLProverManifest, BatchAVLProverSubtree, ProxyInternalNode}
@@ -98,7 +97,7 @@ class VersionedLDBAVLStorage(store: LDBVersionedStore)
     * @param expectedRootHash - expected UTXO set authenticating tree root hash
     * @return - hash of root node of tree, or failure if an error (e.g. in database) happened
     */
-  def dumpSnapshot(dumpStorage: LDBKVStore, manifestDepth: Int, expectedRootHash: Array[Byte]): Try[Array[Byte]] = {
+  def dumpSnapshot(dumpStorage: LDBKVStore, manifestDepth: Byte, expectedRootHash: Array[Byte]): Try[Array[Byte]] = {
     store.processSnapshot { dbReader =>
 
       def subtreeLoop(label: DigestType, builder: mutable.ArrayBuilder[Byte]): Unit = {
@@ -137,14 +136,14 @@ class VersionedLDBAVLStorage(store: LDBVersionedStore)
       }
 
       val rootNodeLabel = dbReader.get(topNodeHashKey)
-      val rootNodeHeight = Ints.fromByteArray(dbReader.get(topNodeHeightKey))
+      val rootNodeHeight = Ints.fromByteArray(dbReader.get(topNodeHeightKey)).toByte
 
       require(rootNodeLabel.sameElements(expectedRootHash), "Root node hash changed")
 
       val manifestBuilder = mutable.ArrayBuilder.make[Byte]()
       manifestBuilder.sizeHint(200000)
-      manifestBuilder ++= Ints.toByteArray(rootNodeHeight)
-      manifestBuilder += ManifestSerializer.MainnetManifestDepth
+      manifestBuilder += rootNodeHeight
+      manifestBuilder += manifestDepth
 
       manifestLoop(rootNodeLabel, level = 1, manifestBuilder)
       val manifestBytes = manifestBuilder.result()
@@ -232,7 +231,7 @@ object VersionedLDBAVLStorage {
     * Calculate tree digest, given root node label(hash) and root node height, by appending height to the hash
     */
   def digest[D <: hash.Digest](rootNodeLabel: D, rootNodeHeight: Int): ADDigest = {
-    assert(rootNodeHeight >= 0 && rootNodeHeight < 256)
+    assert(rootNodeHeight >= 0 && rootNodeHeight <= 127)
     // rootNodeHeight should never be more than 255, so the toByte conversion is safe (though it may cause an incorrect
     // sign on the signed byte if rootHeight>127, but we handle that case correctly on decoding the byte back to int in the
     // verifier, by adding 256 if it's negative).
@@ -241,4 +240,11 @@ object VersionedLDBAVLStorage {
     ADDigest @@ (rootNodeLabel :+ rootNodeHeight.toByte)
   }
 
+  /**
+    * splits 33-byte digest into 32-bytes hash and 1-byte tree height
+    */
+  def splitDigest(digest: ADDigest): (Array[Byte], Byte) = {
+    digest.dropRight(1) -> digest.takeRight(1).head
+  }
+
 }

avldb/src/main/scala/scorex/db/LDBKVStore.scala

@@ -13,6 +13,7 @@ import spire.syntax.all.cfor
   * Both keys and values are var-sized byte arrays.
   */
 class LDBKVStore(protected val db: DB) extends KVStoreReader with ScorexLogging {
+  private val emptyArrayOfByteArray = Array.empty[Array[Byte]]
 
   /**
     * Update this database atomically with a batch of insertion and removal operations
@@ -55,15 +56,16 @@ class LDBKVStore(protected val db: DB) extends KVStoreReader with ScorexLogging
   /**
     * `update` variant where we only insert values into this database
     */
-  def insert(keys: Array[K], values: Array[V]): Try[Unit] = update(keys, values, Array.empty)
-
-  def insert(values: Array[(K, V)]): Try[Unit] = update(values.map(_._1), values.map(_._2), Array.empty)
-
+  def insert(keys: Array[K], values: Array[V]): Try[Unit] = {
+    update(keys, values, emptyArrayOfByteArray)
+  }
 
   /**
     * `update` variant where we only remove values from this database
     */
-  def remove(keys: Array[K]): Try[Unit] = update(Array.empty, Array.empty, keys)
+  def remove(keys: Array[K]): Try[Unit] = {
+    update(emptyArrayOfByteArray, emptyArrayOfByteArray, keys)
+  }
 
   /**
     * Get last key within some range (inclusive) by used comparator.

benchmarks/src/test/scala/org/ergoplatform/db/LDBStoreBench.scala

@@ -39,7 +39,7 @@ object LDBStoreBench
   val txsWithDbGen: Gen[(Seq[BlockTransactions], LDBKVStore)] = txsGen.map { bts =>
     val toInsert = bts.map(bt => idToBytes(bt.headerId) -> bt.bytes).toArray
     val db = storeLDB()
-    toInsert.grouped(5).foreach(db.insert(_).get)
+    toInsert.grouped(5).foreach(kv => db.insert(kv.map(_._1), kv.map(_._2)).get)
     bts -> storeLDB
   }
 
@@ -55,7 +55,7 @@ object LDBStoreBench
   private def benchWriteLDB(bts: Seq[BlockTransactions]): Unit = {
     val toInsert = bts.map(bt => idToBytes(bt.headerId) -> bt.bytes).toArray
     val db = storeLDB()
-    toInsert.grouped(5).foreach(db.insert(_).get)
+    toInsert.grouped(5).foreach(kv => db.insert(kv.map(_._1), kv.map(_._2)).get)
   }
 
   private def benchReadLDB(bts: Seq[BlockTransactions], db: LDBKVStore): Unit = {

papers/emission.md

@@ -97,12 +97,6 @@ second one is to pay mining fee supposedly (its value can be 0.01 ERG at most)
 
     val correctNftId = EQ(firstTokenId, ByteArrayConstant(reemissionNftId))
 
-    // output of the reemission contract
-    val reemissionOut = ByIndex(Outputs, IntConstant(0))
-
-    // output to pay miner
-    val minerOut = ByIndex(Outputs, IntConstant(1))
-
     // miner's output must have script which is time-locking reward for miner's pubkey
     // box height must be the same as block height
     val correctMinerOutput = AND(

src/main/resources/api/openapi.yaml

@@ -96,6 +96,17 @@ components:
         bytes:
           $ref: '#/components/schemas/HexString'
 
+    SnapshotsInfo:
+      type: object
+      required:
+        - availableManifests
+      properties:
+        availableManifests:
+          description: Map of available manifests height -> manifestId
+          type: array
+          items:
+            type: object
+
     ErgoTransactionOutput:
       type: object
       required:
@@ -5077,6 +5088,28 @@ paths:
               schema:
                 $ref: '#/components/schemas/ApiError'
 
+  /utxo/getSnapshotsInfo:
+    get:
+      summary: Get information about locally stored UTXO snapshots
+      operationId: getSnapshotsInfo
+      tags:
+        - utxo
+      responses:
+        '200':
+          description: A list of saved snapshots
+      content:
+        application/json:
+          schema:
+            $ref: '#/components/schemas/SnapshotsInfo'
+    default:
+      description: Error
+      content:
+        application/json:
+          schema:
+            $ref: '#/components/schemas/ApiError'
+
+
+
   /utxo/genesis:
     get:
       summary: Get genesis boxes (boxes existed before the very first block)

src/main/resources/application.conf

@@ -21,8 +21,18 @@ ergo {
     # otherwise, it could be hard to find proofs around the peers
     blocksToKeep = -1
 
-    # Download and apply UTXO set snapshot and full-blocks after that
-    utxoBootstrap = false
+    utxo {
+        # Download and apply UTXO set snapshot and full-blocks after that
+        utxoBootstrap = false
+
+        # how many utxo set snapshots to store, 0 means that they are not stored at all
+        storingUtxoSnapshots = 0
+
+        # how many utxo set snapshots for a height with same id we need to find in p2p network
+        # in order to start downloading it
+        p2pUtxoSnapshots = 2
+    }
+
 
     # Download PoPoW proof on node bootstrap
     PoPoWBootstrap = false
@@ -106,7 +116,7 @@ ergo {
     # }
     #
     # Before the height given (including it) validation of scripts is missed.
-    # This improving perfomance and memory usage during initial bootstrapping.
+    # This improving performance and memory usage during initial bootstrapping.
     # The node still applying transactions to UTXO set and so checks UTXO set digests for each block.
     # Block at checkpoint height is to be checked against expected one.
     checkpoint = null

src/main/scala/org/ergoplatform/ErgoApp.scala

@@ -59,28 +59,35 @@ class ErgoApp(args: Args) extends ScorexLogging {
     else None
   upnpGateway.foreach(_.addPort(scorexSettings.network.bindAddress.getPort))
 
-  //an address to send to peers
-  private val externalSocketAddress: Option[InetSocketAddress] =
-  scorexSettings.network.declaredAddress orElse {
-    upnpGateway.map(u =>
-      new InetSocketAddress(u.externalAddress, scorexSettings.network.bindAddress.getPort)
-    )
+  // own address to send to peers
+  private val externalSocketAddress: Option[InetSocketAddress] = {
+    scorexSettings.network.declaredAddress orElse {
+      upnpGateway.map(u =>
+        new InetSocketAddress(u.externalAddress, scorexSettings.network.bindAddress.getPort)
+      )
+    }
   }
 
-  private val basicSpecs = {
+  // descriptors of p2p networking protocol messages
+  private val p2pMessageSpecifications = {
     Seq(
       GetPeersSpec,
       new PeersSpec(scorexSettings.network.maxPeerSpecObjects),
+      ErgoSyncInfoMessageSpec,
       InvSpec,
       RequestModifierSpec,
-      ModifiersSpec
+      ModifiersSpec,
+      GetSnapshotsInfoSpec,
+      SnapshotsInfoSpec,
+      GetManifestSpec,
+      ManifestSpec,
+      GetUtxoSnapshotChunkSpec,
+      UtxoSnapshotChunkSpec
     )
   }
 
-  private val additionalMessageSpecs: Seq[MessageSpec[_]] = Seq(ErgoSyncInfoMessageSpec)
-
   private val scorexContext = ScorexContext(
-    messageSpecs        = basicSpecs ++ additionalMessageSpecs,
+    messageSpecs        = p2pMessageSpecifications,
     upnpGateway         = upnpGateway,
     externalNodeAddress = externalSocketAddress
   )
@@ -129,7 +136,14 @@ class ErgoApp(args: Args) extends ScorexLogging {
         InvSpec.messageCode                 -> ergoNodeViewSynchronizerRef,
         RequestModifierSpec.messageCode     -> ergoNodeViewSynchronizerRef,
         ModifiersSpec.messageCode           -> ergoNodeViewSynchronizerRef,
-        ErgoSyncInfoMessageSpec.messageCode -> ergoNodeViewSynchronizerRef
+        ErgoSyncInfoMessageSpec.messageCode -> ergoNodeViewSynchronizerRef,
+        // utxo set snapshot exchange related messages
+        GetSnapshotsInfoSpec.messageCode    -> ergoNodeViewSynchronizerRef,
+        SnapshotsInfoSpec.messageCode       -> ergoNodeViewSynchronizerRef,
+        GetManifestSpec.messageCode         -> ergoNodeViewSynchronizerRef,
+        ManifestSpec.messageCode            -> ergoNodeViewSynchronizerRef,
+        GetUtxoSnapshotChunkSpec.messageCode-> ergoNodeViewSynchronizerRef,
+        UtxoSnapshotChunkSpec.messageCode   -> ergoNodeViewSynchronizerRef
       )
       // Launching PeerSynchronizer actor which is then registering itself at network controller
       if (ergoSettings.scorexSettings.network.peerDiscovery) {

src/main/scala/org/ergoplatform/http/api/ApiCodecs.scala

@@ -30,6 +30,8 @@ import io.circe.syntax._
 import org.ergoplatform.http.api.requests.{CryptoResult, ExecuteRequest, HintExtractionRequest}
 import org.ergoplatform.nodeView.history.extra.ExtraIndexer.getAddress
 import org.ergoplatform.nodeView.history.extra.{BalanceInfo, IndexedErgoBox, IndexedErgoTransaction, IndexedToken}
+import org.ergoplatform.nodeView.state.SnapshotsInfo
+import org.ergoplatform.nodeView.state.UtxoState.ManifestId
 import org.ergoplatform.wallet.interface4j.SecretString
 import scorex.crypto.authds.{LeafData, Side}
 import scorex.crypto.authds.merkle.MerkleProof
@@ -405,6 +407,20 @@ trait ApiCodecs extends JsonCodecs {
     } yield TransactionHintsBag(secretHints.mapValues(HintsBag.apply), publicHints.mapValues(HintsBag.apply))
   }
 
+  implicit val SnapshotInfoEncoder: Encoder[SnapshotsInfo] = { si =>
+    Json.obj(
+      "availableManifests" -> si.availableManifests.map { case (height, manifest) =>
+        height -> manifest
+      }.asJson
+    )
+  }
+
+  implicit val SnapshotInfoDecoder: Decoder[SnapshotsInfo] = { cursor =>
+    for {
+      availableManifests <- Decoder.decodeMap[Int, ManifestId].tryDecode(cursor.downField("availableManifests"))
+    } yield new SnapshotsInfo(availableManifests)
+  }
+
   implicit val transactionSigningRequestEncoder: Encoder[TransactionSigningRequest] = { tsr =>
     Json.obj(
       "tx" -> tsr.unsignedTx.asJson,

src/main/scala/org/ergoplatform/http/api/UtxoApiRoute.scala

@@ -6,7 +6,7 @@ import akka.pattern.ask
 import org.ergoplatform.ErgoBox
 import org.ergoplatform.nodeView.ErgoReadersHolder.{GetReaders, Readers}
 import org.ergoplatform.nodeView.mempool.ErgoMemPoolReader
-import org.ergoplatform.nodeView.state.{ErgoStateReader, UtxoStateReader}
+import org.ergoplatform.nodeView.state.{ErgoStateReader, UtxoSetSnapshotPersistence, UtxoStateReader}
 import org.ergoplatform.wallet.boxes.ErgoBoxSerializer
 import scorex.core.api.http.ApiResponse
 import scorex.core.settings.RESTApiSettings
@@ -24,7 +24,7 @@ case class UtxoApiRoute(readersHolder: ActorRef, override val settings: RESTApiS
     (readersHolder ? GetReaders).mapTo[Readers].map(rs => (rs.s, rs.m))
 
   override val route: Route = pathPrefix("utxo") {
-    byId ~ serializedById ~ genesis ~ withPoolById ~ withPoolSerializedById ~ getBoxesBinaryProof
+    byId ~ serializedById ~ genesis ~ withPoolById ~ withPoolSerializedById ~ getBoxesBinaryProof ~ getSnapshotsInfo
   }
 
   def withPoolById: Route = (get & path("withPool" / "byId" / Segment)) { id =>
@@ -82,4 +82,16 @@ case class UtxoApiRoute(readersHolder: ActorRef, override val settings: RESTApiS
       case _ => None
     })
   }
+
+  /**
+    * Handler for /utxo/getSnapshotsInfo API call which is providing list of
+    * UTXO set snapshots stored locally
+    */
+  def getSnapshotsInfo: Route = (get & path("getSnapshotsInfo")) {
+    ApiResponse(getState.map {
+      case usr: UtxoSetSnapshotPersistence =>
+        Some(usr.getSnapshotInfo())
+      case _ => None
+    })
+  }
 }