diff --git a/.github/workflows/bootstrap-mainnet.yml b/.github/workflows/bootstrap-mainnet.yml
new file mode 100644
index 0000000000..0dca61f4d2
--- /dev/null
+++ b/.github/workflows/bootstrap-mainnet.yml
@@ -0,0 +1,42 @@
+name: Mainnet UTXO snapshot bootstrap test
+
+# Boots a mainnet node with nipopowBootstrap + utxoBootstrap from an empty data dir and
+# waits for a full sync. Validates the fast-bootstrap path end-to-end against the real
+# network. Kept separate from release.yml so the multi-hour run cannot block publishing.
+
+on:
+  release:
+    types: [published]
+  workflow_dispatch:
+
+jobs:
+  bootstrap_mainnet:
+    name: NiPoPoW + UTXO snapshot bootstrap on mainnet
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout current branch (full)
+        uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+
+      - name: Setup Java and Scala
+        uses: olafurpg/setup-scala@v10
+        with:
+          java-version: adopt@1.8
+
+      - name: Cache sbt
+        uses: actions/cache@v4
+        with:
+          path: |
+            ~/.sbt
+            ~/.ivy2/cache
+            ~/.coursier/cache/v1
+            ~/.cache/coursier/v1
+            ~/AppData/Local/Coursier/Cache/v1
+            ~/Library/Caches/Coursier/Cache/v1
+          key: ${{ runner.os }}-sbt-cache-v4-${{ hashFiles('**/*.sbt') }}-${{ hashFiles('project/build.properties') }}
+
+      - name: Run mainnet bootstrap test
+        run: |
+          mkdir tmp
+          TMPDIR=$(pwd)/tmp sbt -Denv=test clean ++2.12.20 docker "it2:testOnly *TestUtxoSnapshotBootstrapOnMainNetSpec"
diff --git a/.gitignore b/.gitignore
index aad6dae921..eb475981a5 100644
--- a/.gitignore
+++ b/.gitignore
@@ -7,6 +7,7 @@ devnet
 .ensime
 .ensime_cache/
 scorex.yaml
+*.iml
 
 # LLM reports on code analysis etc
 llm_generated
diff --git a/src/it2/scala/org/ergoplatform/it2/TestUtxoSnapshotBootstrapOnMainNetSpec.scala b/src/it2/scala/org/ergoplatform/it2/TestUtxoSnapshotBootstrapOnMainNetSpec.scala
new file mode 100644
index 0000000000..6db60d92c2
--- /dev/null
+++ b/src/it2/scala/org/ergoplatform/it2/TestUtxoSnapshotBootstrapOnMainNetSpec.scala
@@ -0,0 +1,63 @@
+package org.ergoplatform.it2
+
+import com.typesafe.config.{Config, ConfigFactory}
+import org.ergoplatform.it.api.NodeApi.NodeInfo
+import org.ergoplatform.it.container.{IntegrationSuite, Node}
+import org.scalatest.OptionValues
+import org.scalatest.flatspec.AnyFlatSpec
+
+import scala.async.Async
+import scala.concurrent.Await
+import scala.concurrent.duration._
+
+class TestUtxoSnapshotBootstrapOnMainNetSpec
+  extends AnyFlatSpec
+    with IntegrationSuite
+    with OptionValues {
+
+  // Unlike TestOnMainNetSpec, no host volume is mounted: node data lives in the container's
+  // anonymous volume (/home/ergo/.ergo) and is discarded with the container, so every run
+  // performs a real bootstrap from an empty data dir.
+
+  val bootstrapConfig: Config = ConfigFactory.parseString(
+    s"""
+       |ergo.node.utxo.utxoBootstrap = true
+       |ergo.node.nipopow.nipopowBootstrap = true
+       |# genesisId of mainnet.conf, needed for ErgoSettings validation on the test host,
+       |# where the network config file is not loaded (see Docker.buildErgoSettings)
+       |ergo.chain.genesisId = "b0244dfc267baca974a4caee06120321562784303a8a688976ae56170e4d175b"
+    """.stripMargin
+  )
+
+  val nodeConfig: Config = bootstrapConfig
+    .withFallback(nodeSeedConfigs.head)
+    .withFallback(nonGeneratingPeerConfig)
+  val node: Node = docker.startMainNetNodeYesImSure(nodeConfig).get
+
+  it should "Bootstrap from a UTXO set snapshot via NiPoPoW proof on mainnet and fully sync" in {
+    // Phase 1: headers appear, proving the trusted NiPoPoW proof was applied
+    val headersResult = Async.async {
+      Async.await(node.waitFor[NodeInfo](
+        _.info,
+        nodeInfo => nodeInfo.bestHeaderHeightOpt.exists(_ > 1000),
+        1.minute
+      ))
+    }
+    val nodeInfoAfterHeaders = Await.result(headersResult, 1.hour)
+    log.info(s"Headers appeared, best header height: ${nodeInfoAfterHeaders.bestHeaderHeightOpt}")
+
+    // Phase 2: wait for a full sync (snapshot applied + full blocks downloaded)
+    val syncResult = Async.async {
+      Async.await(node.waitFor[NodeInfo](
+        _.info,
+        nodeInfo => nodeInfo.bestBlockHeightOpt.exists(nodeInfo.bestHeaderHeightOpt.contains),
+        1.minute
+      ))
+    }
+    val syncedInfo = Await.result(syncResult, 5.hours)
+
+    // guard against a degenerate "synced at genesis" pass
+    syncedInfo.bestHeaderHeightOpt.value should be > 1000000
+  }
+
+}
diff --git a/src/main/scala/org/ergoplatform/network/ErgoNodeViewSynchronizer.scala b/src/main/scala/org/ergoplatform/network/ErgoNodeViewSynchronizer.scala
index ced63d1e8a..021898dc87 100644
--- a/src/main/scala/org/ergoplatform/network/ErgoNodeViewSynchronizer.scala
+++ b/src/main/scala/org/ergoplatform/network/ErgoNodeViewSynchronizer.scala
@@ -1147,8 +1147,22 @@ class ErgoNodeViewSynchronizer(networkControllerRef: ActorRef,
           Seq.empty
         }
       case _ =>
-        log.info(s"Processing ${invData.ids.length} non-tx invs (of type $modifierTypeId) from $peer")
-        invData.ids.filter(mid => deliveryTracker.status(mid, modifierTypeId, Seq(hr)) == ModifiersStatus.Unknown)
+        // During UTXO set snapshot bootstrap, ignore block-section invs (extension, transactions, ADProofs)
+        // until the snapshot is applied. Headers and snapshot-related types are still processed.
+        val utxoBootstrapInProgress =
+          settings.nodeSettings.utxoSettings.utxoBootstrap &&
+            !hr.isUtxoSnapshotApplied &&
+            modifierTypeId != Header.modifierTypeId &&
+            modifierTypeId != ManifestTypeId.value &&
+            modifierTypeId != UtxoSnapshotChunkTypeId.value
+
+        if (utxoBootstrapInProgress) {
+          log.debug(s"Ignoring ${invData.ids.length} non-tx invs (of type $modifierTypeId) from $peer: UTXO snapshot bootstrap in progress")
+          Seq.empty
+        } else {
+          log.info(s"Processing ${invData.ids.length} non-tx invs (of type $modifierTypeId) from $peer")
+          invData.ids.filter(mid => deliveryTracker.status(mid, modifierTypeId, Seq(hr)) == ModifiersStatus.Unknown)
+        }
     }
 
     if (newModifierIds.nonEmpty) {
diff --git a/src/main/scala/org/ergoplatform/nodeView/ErgoNodeViewHolder.scala b/src/main/scala/org/ergoplatform/nodeView/ErgoNodeViewHolder.scala
index 0363d7bd26..92fc48ebaf 100644
--- a/src/main/scala/org/ergoplatform/nodeView/ErgoNodeViewHolder.scala
+++ b/src/main/scala/org/ergoplatform/nodeView/ErgoNodeViewHolder.scala
@@ -316,6 +316,12 @@ abstract class ErgoNodeViewHolder[State <: ErgoState[State]](settings: ErgoSetti
       if (history().isEmpty) {
         history().applyPopowProof(proof)
         if (!history().isEmpty) {
+          // When UTXO set snapshot bootstrap is enabled, mark headers chain as synced right after
+          // a trusted NiPoPoW proof is applied. This allows the node to start requesting UTXO set
+          // snapshots immediately, instead of waiting for normal header sync to reach the tip.
+          if (settings.nodeSettings.utxoSettings.utxoBootstrap) {
+            history().setHeadersChainSynced()
+          }
           updateNodeView(updatedHistory = Some(history()))
         }
       }
@@ -583,6 +589,18 @@ abstract class ErgoNodeViewHolder[State <: ErgoState[State]](settings: ErgoSetti
       )
   }
 
+  private def isPreparedUtxoSnapshotState(state: State, history: ErgoHistory): Boolean =
+    ErgoNodeViewHolder.isPreparedUtxoSnapshotState(
+      state.isInstanceOf[UtxoState],
+      settings.nodeSettings.utxoSettings.utxoBootstrap,
+      history.isUtxoSnapshotApplied,
+      state.version,
+      state.rootDigest,
+      {
+        val snapshotHeight = history.minimalFullBlockHeight - 1
+        history.bestHeaderAtHeight(snapshotHeight)
+      })
+
   private def restoreConsistentState(stateIn: State, history: ErgoHistory): Try[State] = {
     (stateIn.version, history.bestFullBlockOpt, stateIn) match {
       case (ErgoState.genesisStateVersion, None, _) =>
@@ -591,8 +609,12 @@ abstract class ErgoNodeViewHolder[State <: ErgoState[State]](settings: ErgoSetti
       case (stateId, Some(block), _) if stateId == block.id =>
         log.info(s"State and history have the same version ${encoder.encode(stateId)}, no recovery needed.")
         Success(stateIn)
+      case (_, None, _) if isPreparedUtxoSnapshotState(stateIn, history) =>
+        log.info(s"Prepared UTXO snapshot state ${encoder.encode(stateIn.version)} restored before the first full block")
+        Success(stateIn)
       case (_, None, _) =>
         log.info("State and history are inconsistent. History is empty on startup, rollback state to genesis.")
+        stateIn.closeStorage()
         Success(recreatedState())
       case (_, Some(bestFullBlock), _: DigestState) =>
         log.info(s"State and history are inconsistent. Going to switch state to version ${bestFullBlock.encodedId}")
@@ -722,6 +744,25 @@ abstract class ErgoNodeViewHolder[State <: ErgoState[State]](settings: ErgoSetti
 
 object ErgoNodeViewHolder {
 
+  private[nodeView] def isPreparedUtxoSnapshotState(
+      stateIsUtxo: Boolean,
+      utxoBootstrap: => Boolean,
+      snapshotApplied: => Boolean,
+      stateVersion: VersionTag,
+      stateRoot: Array[Byte],
+      snapshotHeaderOpt: => Option[Header]): Boolean =
+    stateIsUtxo &&
+      utxoBootstrap &&
+      snapshotApplied &&
+      snapshotHeaderOpt.exists(matchesPreparedUtxoSnapshotHeader(stateVersion, stateRoot, _))
+
+  private def matchesPreparedUtxoSnapshotHeader(
+      stateVersion: VersionTag,
+      stateRoot: Array[Byte],
+      header: Header): Boolean =
+    stateVersion == idToVersion(header.id) &&
+      java.util.Arrays.equals(stateRoot, header.stateRoot)
+
   object ReceivableMessages {
     // Tracking last modifier and header & block heights in time, being periodically checked for possible stuck
     case class ChainProgress(lastMod: BlockSection, headersHeight: Int, blockHeight: Int, lastUpdate: Long)
diff --git a/src/main/scala/org/ergoplatform/nodeView/history/storage/modifierprocessors/FullBlockPruningProcessor.scala b/src/main/scala/org/ergoplatform/nodeView/history/storage/modifierprocessors/FullBlockPruningProcessor.scala
index 95de235c49..e0e91811d3 100644
--- a/src/main/scala/org/ergoplatform/nodeView/history/storage/modifierprocessors/FullBlockPruningProcessor.scala
+++ b/src/main/scala/org/ergoplatform/nodeView/history/storage/modifierprocessors/FullBlockPruningProcessor.scala
@@ -30,6 +30,14 @@ trait FullBlockPruningProcessor extends MinimalFullBlockHeightFunctions {
     */
   def isHeadersChainSynced: Boolean = isHeadersChainSyncedVar
 
+  /** Marks headers chain as synchronized. Used by NiPoPoW / UTXO-set-snapshot bootstrap to start
+    * downloading UTXO set snapshots immediately after a trusted NiPoPoW proof is applied,
+    * instead of waiting for a "new" header to arrive through normal header sync.
+    */
+  def setHeadersChainSynced(): Unit = {
+    if (!isHeadersChainSyncedVar) isHeadersChainSyncedVar = true
+  }
+
   /** Start height to download full blocks from
     */
   def minimalFullBlockHeight: Int = readMinimalFullBlockHeight()
diff --git a/src/main/scala/org/ergoplatform/nodeView/history/storage/modifierprocessors/ToDownloadProcessor.scala b/src/main/scala/org/ergoplatform/nodeView/history/storage/modifierprocessors/ToDownloadProcessor.scala
index 090610c3c2..bb003769da 100644
--- a/src/main/scala/org/ergoplatform/nodeView/history/storage/modifierprocessors/ToDownloadProcessor.scala
+++ b/src/main/scala/org/ergoplatform/nodeView/history/storage/modifierprocessors/ToDownloadProcessor.scala
@@ -111,6 +111,11 @@ trait ToDownloadProcessor
     if (!nodeSettings.verifyTransactions) {
       // A regime that do not download and verify transaction
       Nil
+    } else if (nodeSettings.utxoSettings.utxoBootstrap && !isUtxoSnapshotApplied) {
+      // While bootstrapping from a UTXO set snapshot, do not download full block sections
+      // until the snapshot has been applied. Block sections downloaded before the snapshot
+      // would be stored as non-best and never applied to the freshly recreated state.
+      Nil
     } else if (shouldDownloadBlockAtHeight(header.height)) {
       // Already synced and header is not too far back. Download required modifiers.
       requiredModifiersForHeader(header)
diff --git a/src/test/scala/org/ergoplatform/network/ErgoNodeViewSynchronizerSpecification.scala b/src/test/scala/org/ergoplatform/network/ErgoNodeViewSynchronizerSpecification.scala
index 3e82649f70..a8998129e3 100644
--- a/src/test/scala/org/ergoplatform/network/ErgoNodeViewSynchronizerSpecification.scala
+++ b/src/test/scala/org/ergoplatform/network/ErgoNodeViewSynchronizerSpecification.scala
@@ -3,7 +3,8 @@ package org.ergoplatform.network
 import akka.actor.{ActorRef, ActorSystem, Cancellable, Props}
 import akka.testkit.TestProbe
 import org.ergoplatform.modifiers.history.header.{Header, HeaderSerializer}
-import org.ergoplatform.modifiers.{BlockSection, ErgoFullBlock}
+import org.ergoplatform.modifiers.history.extension.Extension
+import org.ergoplatform.modifiers.{BlockSection, ErgoFullBlock, ManifestTypeId, UtxoSnapshotChunkTypeId}
 import org.ergoplatform.network.ErgoNodeViewSynchronizerMessages._
 import org.ergoplatform.nodeView.ErgoNodeViewHolder
 import org.ergoplatform.nodeView.history.{ErgoHistory, ErgoHistoryReader, ErgoSyncInfoMessageSpec, ErgoSyncInfoV2}
@@ -11,7 +12,7 @@ import org.ergoplatform.nodeView.mempool.ErgoMemPool
 import org.ergoplatform.nodeView.state.wrapped.WrappedUtxoState
 import org.ergoplatform.nodeView.state.{StateType, UtxoState}
 import org.ergoplatform.sanity.ErgoSanity._
-import org.ergoplatform.settings.{ErgoSettings, ErgoSettingsReader}
+import org.ergoplatform.settings.{ErgoSettings, ErgoSettingsReader, UtxoSettings}
 import org.ergoplatform.validation.{ParentHeaderNotFoundError, RecoverableModifierError}
 import org.ergoplatform.wallet.utils.FileUtils
 import org.scalacheck.Gen
@@ -41,6 +42,7 @@ class ErgoNodeViewSynchronizerSpecification extends AnyPropSpec
   import org.ergoplatform.utils.ErgoCoreTestConstants._
   import org.ergoplatform.utils.generators.ErgoNodeTransactionGenerators._
   import org.ergoplatform.utils.generators.ConnectedPeerGenerators._
+  import org.ergoplatform.utils.generators.CoreObjectGenerators._
   import org.ergoplatform.utils.generators.ErgoCoreTransactionGenerators._
   import org.ergoplatform.utils.generators.ValidBlocksGenerators._
   import org.ergoplatform.utils.generators.ChainGenerator._
@@ -203,6 +205,77 @@ class ErgoNodeViewSynchronizerSpecification extends AnyPropSpec
     )
   }
 
+  /**
+    * Fixture for UTXO set snapshot bootstrap tests: synchronizer and history are built with
+    * `utxoBootstrap` enabled (or disabled, for control tests), history contains headers only.
+    */
+  class UtxoBootstrapSynchronizerFixture(utxoBootstrap: Boolean) extends AkkaFixture {
+    implicit val ec: ExecutionContextExecutor = system.dispatcher
+    val ncProbe = TestProbe("NetworkControllerProbe")
+    val pchProbe = TestProbe("PeerHandlerProbe")
+    val syncTracker = ErgoSyncTracker(settings.scorexSettings.network)
+    val deliveryTracker: DeliveryTracker = DeliveryTracker.empty(settings)
+
+    val synchronizerSettings = settings.copy(
+      nodeSettings = settings.nodeSettings.copy(
+        utxoSettings = UtxoSettings(utxoBootstrap, 0, 2)
+      )
+    )
+
+    deleteRecursive(ErgoHistory.historyDir(synchronizerSettings))
+    val nodeViewHolderMockRef = system.actorOf(Props(new NodeViewHolderMock))
+
+    val synchronizerMockRef = system.actorOf(Props(
+      new SynchronizerMock(
+        ncProbe.ref,
+        nodeViewHolderMockRef,
+        ErgoSyncInfoMessageSpec,
+        synchronizerSettings,
+        syncTracker,
+        deliveryTracker)
+    ))
+
+    val history = generateHistory(verifyTransactions = true,
+                                  StateType.Utxo,
+                                  PoPoWBootstrap = false,
+                                  blocksToKeep = -1,
+                                  utxoBootstrap = utxoBootstrap)
+    val chain = genHeaderChain(BlocksInChain, history, diffBitsOpt = None, useRealTs = false)
+    val updHistory = applyHeaderChain(history, chain)
+
+    synchronizerMockRef ! ChangedHistory(updHistory)
+    synchronizerMockRef ! ChangedMempool(ErgoMemPool.empty(synchronizerSettings))
+
+    val peerInfo = PeerInfo(defaultPeerSpec, System.currentTimeMillis())
+    @SuppressWarnings(Array("org.wartremover.warts.OptionPartial"))
+    val peer: ConnectedPeer = ConnectedPeer(
+      connectionIdGen.sample.get,
+      pchProbe.ref,
+      Some(peerInfo)
+    )
+  }
+
+  private def withUtxoBootstrapFixture(utxoBootstrap: Boolean)(testCode: UtxoBootstrapSynchronizerFixture => Any): Unit = {
+    val fixture = new UtxoBootstrapSynchronizerFixture(utxoBootstrap)
+    try {
+      testCode(fixture)
+    }
+    finally {
+      Await.result(fixture.system.terminate(), Duration.Inf)
+    }
+  }
+
+  private def requestForModifierSent(ncProbe: TestProbe, typeId: org.ergoplatform.modifiers.NetworkObjectTypeId.Value, id: scorex.util.ModifierId): Unit = {
+    ncProbe.fishForMessage(3 seconds) {
+      case stn: SendToNetwork =>
+        stn.message.spec.messageCode == RequestModifierSpec.messageCode && {
+          val invData = stn.message.data.get.asInstanceOf[InvData]
+          invData.typeId == typeId && invData.ids.contains(id)
+        }
+      case _ => false
+    }
+  }
+
   property("NodeViewSynchronizer: Message: SyncInfoSpec V2 - younger peer") {
     withFixture { ctx =>
       import ctx._
@@ -1168,4 +1241,81 @@ class ErgoNodeViewSynchronizerSpecification extends AnyPropSpec
     }
   }
 
+  /**
+    * During UTXO set snapshot bootstrap block section invs must be ignored until the snapshot is applied.
+    */
+  property("NodeViewSynchronizer: InvSpec - block section invs are ignored during utxo bootstrap") {
+    withUtxoBootstrapFixture(utxoBootstrap = true) { ctx =>
+      import ctx._
+
+      val unknownSectionId = modifierIdGen.sample.get
+      synchronizerMockRef ! Message(InvSpec,
+        Left(InvSpec.toBytes(InvData(Extension.modifierTypeId, Seq(unknownSectionId)))),
+        Some(peer))
+
+      // request must not be sent, the inv is dropped
+      ncProbe.expectNoMessage(1.second)
+    }
+  }
+
+  property("NodeViewSynchronizer: InvSpec - header invs are still processed during utxo bootstrap") {
+    withUtxoBootstrapFixture(utxoBootstrap = true) { ctx =>
+      import ctx._
+
+      val unknownHeaderId = modifierIdGen.sample.get
+      synchronizerMockRef ! Message(InvSpec,
+        Left(InvSpec.toBytes(InvData(Header.modifierTypeId, Seq(unknownHeaderId)))),
+        Some(peer))
+
+      requestForModifierSent(ncProbe, Header.modifierTypeId, unknownHeaderId)
+    }
+  }
+
+  property("NodeViewSynchronizer: InvSpec - snapshot-related invs are not filtered during utxo bootstrap") {
+    withUtxoBootstrapFixture(utxoBootstrap = true) { ctx =>
+      import ctx._
+
+      val manifestId = modifierIdGen.sample.get
+      synchronizerMockRef ! Message(InvSpec,
+        Left(InvSpec.toBytes(InvData(ManifestTypeId.value, Seq(manifestId)))),
+        Some(peer))
+      requestForModifierSent(ncProbe, ManifestTypeId.value, manifestId)
+
+      val chunkId = modifierIdGen.sample.get
+      synchronizerMockRef ! Message(InvSpec,
+        Left(InvSpec.toBytes(InvData(UtxoSnapshotChunkTypeId.value, Seq(chunkId)))),
+        Some(peer))
+      requestForModifierSent(ncProbe, UtxoSnapshotChunkTypeId.value, chunkId)
+    }
+  }
+
+  property("NodeViewSynchronizer: InvSpec - block section invs processed after utxo snapshot applied") {
+    withUtxoBootstrapFixture(utxoBootstrap = true) { ctx =>
+      import ctx._
+
+      // apply snapshot, so that block sections downloading is allowed again
+      updHistory.onUtxoSnapshotApplied(ctx.chain.last.height)
+
+      val unknownSectionId = modifierIdGen.sample.get
+      synchronizerMockRef ! Message(InvSpec,
+        Left(InvSpec.toBytes(InvData(Extension.modifierTypeId, Seq(unknownSectionId)))),
+        Some(peer))
+
+      requestForModifierSent(ncProbe, Extension.modifierTypeId, unknownSectionId)
+    }
+  }
+
+  property("NodeViewSynchronizer: InvSpec - block section invs processed when utxoBootstrap disabled") {
+    withUtxoBootstrapFixture(utxoBootstrap = false) { ctx =>
+      import ctx._
+
+      val unknownSectionId = modifierIdGen.sample.get
+      synchronizerMockRef ! Message(InvSpec,
+        Left(InvSpec.toBytes(InvData(Extension.modifierTypeId, Seq(unknownSectionId)))),
+        Some(peer))
+
+      requestForModifierSent(ncProbe, Extension.modifierTypeId, unknownSectionId)
+    }
+  }
+
 }
diff --git a/src/test/scala/org/ergoplatform/nodeView/history/storage/modifierprocessors/UtxoBootstrapToDownloadSpecification.scala b/src/test/scala/org/ergoplatform/nodeView/history/storage/modifierprocessors/UtxoBootstrapToDownloadSpecification.scala
new file mode 100644
index 0000000000..a601a79a22
--- /dev/null
+++ b/src/test/scala/org/ergoplatform/nodeView/history/storage/modifierprocessors/UtxoBootstrapToDownloadSpecification.scala
@@ -0,0 +1,133 @@
+package org.ergoplatform.nodeView.history.storage.modifierprocessors
+
+import org.ergoplatform.modifiers.SnapshotsInfoTypeId
+import org.ergoplatform.modifiers.history.HeaderChain
+import org.ergoplatform.nodeView.history.ErgoHistory
+import org.ergoplatform.nodeView.state.StateType
+import org.ergoplatform.serialization.ManifestSerializer
+import org.ergoplatform.utils.ErgoCorePropertyTest
+
+/**
+  * Tests for UTXO set snapshot bootstrap behavior in `ToDownloadProcessor` /
+  * `FullBlockPruningProcessor`:
+  * - `setHeadersChainSynced` makes `nextModifiersToDownload` issue a UTXO set snapshot request
+  *   (instead of full blocks) when no full blocks are applied yet
+  * - `toDownload` returns no block sections until a UTXO set snapshot is applied
+  * - without `utxoBootstrap` enabled, none of the above holds
+  */
+class UtxoBootstrapToDownloadSpecification extends ErgoCorePropertyTest {
+  import org.ergoplatform.utils.HistoryTestHelpers._
+  import org.ergoplatform.utils.ErgoCoreTestConstants._
+  import org.ergoplatform.utils.generators.ChainGenerator._
+  import org.ergoplatform.utils.generators.ErgoNodeTransactionGenerators._
+  import org.ergoplatform.utils.generators.ValidBlocksGenerators._
+
+  private def genUtxoBootstrapHistory() =
+    generateHistory(verifyTransactions = true,
+                    StateType.Utxo,
+                    PoPoWBootstrap = false,
+                    BlocksToKeep,
+                    utxoBootstrap = true)
+
+  private def headersWithFreshTail(history: ErgoHistory, extra: Int = 1) = {
+    val chain = genChain(BlocksInChain + extra, history)
+    val headers = HeaderChain(chain.dropRight(extra).map(_.header))
+    val updHistory = applyHeaderChain(history, headers)
+    (updHistory, chain)
+  }
+
+  property("snapshot request is issued when headers chain synced and no full blocks applied") {
+    var history = genUtxoBootstrapHistory()
+    val chain = genChain(BlocksInChain, history)
+    history = applyHeaderChain(history, HeaderChain(chain.map(_.header)))
+
+    history.bestFullBlockOpt shouldBe None
+    history.isHeadersChainSynced shouldBe false
+
+    // nothing to download before headers chain is considered synced
+    history.nextModifiersToDownload(1, (_, id) => !history.contains(id)) shouldBe
+      Map.empty
+
+    history.setHeadersChainSynced()
+    history.isHeadersChainSynced shouldBe true
+
+    // no full blocks applied, no snapshot plan yet => ask peers for UTXO set snapshots
+    history.nextModifiersToDownload(1, (_, id) => !history.contains(id)) shouldBe
+      Map(SnapshotsInfoTypeId.value -> Seq.empty)
+
+    // setter must be idempotent
+    history.setHeadersChainSynced()
+    history.isHeadersChainSynced shouldBe true
+  }
+
+  property("no repeated snapshot request once download plan is registered") {
+    var history = genUtxoBootstrapHistory()
+    val (updHistory, chain) = headersWithFreshTail(history)
+    history = updHistory
+    history.setHeadersChainSynced()
+    val freshHeader = chain.last.header
+
+    // manifest of some UTXO set snapshot, needed to create a download plan
+    val bh = boxesHolderGenOfSize(1024).sample.get
+    val us = createUtxoState(bh, parameters)
+    val snapshotHeight = freshHeader.height
+    us.dumpSnapshot(snapshotHeight, us.rootDigest.dropRight(1))
+    val manifestId = us.snapshotsDb.readSnapshotsInfo.availableManifests(snapshotHeight)
+    val manifestBytes = us.snapshotsDb.readManifestBytes(manifestId).get
+    val manifest = ManifestSerializer.defaultSerializer.parseBytes(manifestBytes)
+
+    history.registerManifestToDownload(manifest, snapshotHeight, Seq.empty)
+    history.utxoSetSnapshotDownloadPlan() should not be empty
+    history.isUtxoSnapshotApplied shouldBe false
+
+    // download plan exists, so no new snapshot info request
+    history.nextModifiersToDownload(1, (_, id) => !history.contains(id)) shouldBe
+      Map.empty
+  }
+
+  property("toDownload returns no block sections before snapshot, and sections after snapshot") {
+    var history = genUtxoBootstrapHistory()
+    val (updHistory, chain) = headersWithFreshTail(history, extra = 2)
+    history = updHistory
+    history.setHeadersChainSynced()
+    val freshHeader = chain(chain.length - 2).header
+    val nextHeader = chain.last.header
+
+    // headers chain is synced and the header is not too far back, still no block sections
+    // must be downloaded before the UTXO set snapshot is applied
+    val piBefore = history.append(freshHeader).get._2
+    piBefore.toDownload shouldBe Seq.empty
+
+    // apply snapshot at freshHeader's height, so that full blocks downloading
+    // starts from nextHeader
+    history.onUtxoSnapshotApplied(freshHeader.height)
+    history.isUtxoSnapshotApplied shouldBe true
+
+    val piAfter = history.append(nextHeader).get._2
+    piAfter.toDownload shouldBe history.requiredModifiersForHeader(nextHeader)
+    piAfter.toDownload should not be empty
+  }
+
+  property("without utxoBootstrap no snapshot request and block sections downloaded as usual") {
+    var history =
+      generateHistory(verifyTransactions = true,
+                      StateType.Utxo,
+                      PoPoWBootstrap = false,
+                      BlocksToKeep)
+    val (updHistory, chain) = headersWithFreshTail(history)
+    history = updHistory
+    history.setHeadersChainSynced()
+    val freshHeader = chain.last.header
+
+    // full block sections are requested right away, no snapshot request is involved
+    val pi = history.append(freshHeader).get._2
+    pi.toDownload shouldBe history.requiredModifiersForHeader(freshHeader)
+    pi.toDownload should not be empty
+
+    val toDownloadMap =
+      history.nextModifiersToDownload(1, (_, id) => !history.contains(id))
+    toDownloadMap should not be empty
+    toDownloadMap.contains(SnapshotsInfoTypeId.value) shouldBe false
+  }
+
+}
diff --git a/src/test/scala/org/ergoplatform/nodeView/viewholder/ErgoNodeViewHolderSpec.scala b/src/test/scala/org/ergoplatform/nodeView/viewholder/ErgoNodeViewHolderSpec.scala
index 2d8e8f4359..fbf7c1d300 100644
--- a/src/test/scala/org/ergoplatform/nodeView/viewholder/ErgoNodeViewHolderSpec.scala
+++ b/src/test/scala/org/ergoplatform/nodeView/viewholder/ErgoNodeViewHolderSpec.scala
@@ -1,8 +1,9 @@
 package org.ergoplatform.nodeView.viewholder
 
 import java.io.File
+import org.ergoplatform.core.idToVersion
 import org.ergoplatform.ErgoBoxCandidate
-import org.ergoplatform.modifiers.ErgoFullBlock
+import org.ergoplatform.modifiers.{ErgoFullBlock, SnapshotsInfoTypeId}
 import org.ergoplatform.modifiers.history.BlockTransactions
 import org.ergoplatform.modifiers.history.header.Header
 import org.ergoplatform.modifiers.history.popow.NipopowAlgos
@@ -14,6 +15,7 @@ import org.ergoplatform.nodeView.state._
 import org.ergoplatform.nodeView.state.wrapped.WrappedUtxoState
 import org.ergoplatform.settings.{Algos, ErgoSettings}
 import org.ergoplatform.utils.{ErgoCorePropertyTest, NodeViewTestConfig, NodeViewTestOps, RandomWrapper, TestCase}
+import org.ergoplatform.utils.fixtures.NodeViewFixture
 import org.ergoplatform.validation.MalformedModifierError
 import org.ergoplatform.nodeView.ErgoNodeViewHolder.ReceivableMessages._
 import org.ergoplatform.network.ErgoNodeViewSynchronizerMessages._
@@ -32,6 +34,7 @@ class ErgoNodeViewHolderSpec extends ErgoCorePropertyTest with NodeViewTestOps w
   import org.ergoplatform.utils.generators.CoreObjectGenerators._
   import org.ergoplatform.utils.HistoryTestHelpers._
   import org.ergoplatform.utils.generators.ValidBlocksGenerators._
+  import org.ergoplatform.utils.generators.ChainGenerator._
 
   private val t0 = TestCase("check chain is healthy") { fixture =>
     val (us, bh) = createUtxoState(settings)
@@ -618,6 +621,47 @@ class ErgoNodeViewHolderSpec extends ErgoCorePropertyTest with NodeViewTestOps w
     }
   }
 
+  /**
+    * Applies a valid NiPoPoW proof from a separately generated chain to an empty node view holder.
+    * With utxoBootstrap enabled, the node must start UTXO set snapshot bootstrap right after the proof
+    * (headers chain marked as synced, no full blocks downloaded yet). Without utxoBootstrap, normal
+    * full blocks downloading must be started instead.
+    */
+  private val t22 = TestCase("apply nipopow proof to empty holder") { fixture =>
+    import fixture._
+
+    // sender history: generate a chain and a NiPoPoW proof for it
+    val senderHistory = generateHistory(verifyTransactions = true, StateType.Utxo, PoPoWBootstrap = false, blocksToKeep = -1)
+    val senderChain = genChain(5000, senderHistory)
+    val updSenderHistory = applyChain(senderHistory, senderChain)
+    val popowProof = updSenderHistory.nipopowSerializer.parseBytes(updSenderHistory.popowProofBytes().get)
+
+    // the holder must expect sender's genesis id, as with nipopow bootstrapping
+    updateConfig(genesisIdConfig(updSenderHistory.bestHeaderAtHeight(1).map(_.id)))
+
+    subscribeEvents(classOf[ChangedHistory])
+
+    nodeViewHolderRef ! ProcessNipopow(popowProof)
+    expectMsgType[ChangedHistory]
+
+    getHistory.headersHeight shouldBe updSenderHistory.headersHeight
+    getHistory.isHeadersChainSynced shouldBe true
+
+    val toDownloadMap = getHistory.nextModifiersToDownload(1, (_, id) => !getHistory.contains(id))
+    if (settings.nodeSettings.utxoSettings.utxoBootstrap) {
+      // no full blocks must be downloaded before UTXO set snapshot is applied, ask peers for snapshots
+      toDownloadMap shouldBe Map(SnapshotsInfoTypeId.value -> Seq.empty)
+    } else {
+      // normal nipopow bootstrap: full blocks downloading is started, no snapshot request
+      toDownloadMap.contains(SnapshotsInfoTypeId.value) shouldBe false
+    }
+
+    // second proof must not be applied as history is not empty anymore
+    nodeViewHolderRef ! ProcessNipopow(popowProof)
+    expectNoMsg()
+    getHistory.headersHeight shouldBe updSenderHistory.headersHeight
+  }
+
   val cases: List[TestCase] = List(t0, t1, t2, t3, t3a, t4, t5, t6, t7, t8, t9)
 
   NodeViewTestConfig.allConfigs.foreach { c =>
@@ -638,6 +682,187 @@ class ErgoNodeViewHolderSpec extends ErgoCorePropertyTest with NodeViewTestOps w
     }
   }
 
+  property("preserve a prepared UTXO snapshot state across restart") {
+    val protoSettings = NodeViewTestConfig(StateType.Utxo, verifyTransactions = true, popowBootstrap = false)
+      .toSettings
+    val snapshotSettings = protoSettings.copy(
+      nodeSettings = protoSettings.nodeSettings.copy(
+        utxoSettings = protoSettings.nodeSettings.utxoSettings.copy(utxoBootstrap = true)
+      )
+    )
+
+    new NodeViewFixture(snapshotSettings, parameters).apply { fixture =>
+      import fixture._
+
+      val (sourceState, boxHolder) = createUtxoState(settings)
+      val snapshotBlock = validFullBlock(None, sourceState, boxHolder)
+      val sourceAtSnapshot = WrappedUtxoState(sourceState, boxHolder, settings)
+        .applyModifier(snapshotBlock)(_ => ())
+        .get
+      val nextBlock = validFullBlock(Some(snapshotBlock), sourceAtSnapshot)
+
+      applyHeader(snapshotBlock.header).get
+      getHistory.onUtxoSnapshotApplied(snapshotBlock.height)
+      stopNodeViewHolder()
+
+      val stateDir = new File(s"${nodeViewDir.getAbsolutePath}/state")
+      fixture.deleteRecursive(stateDir)
+      stateDir.mkdirs() shouldBe true
+      val persistedGenesis = ErgoState
+        .generateGenesisUtxoState(stateDir, settings, Some(parameters))
+        ._1
+      val persistedSnapshot = persistedGenesis
+        .applyModifier(snapshotBlock, None)(_ => ())
+        .get
+      persistedSnapshot.closeStorage()
+
+      startNodeViewHolder()
+
+      getRootHash shouldBe Algos.encode(snapshotBlock.header.stateRoot)
+      applyBlock(nextBlock) shouldBe 'success
+      getRootHash shouldBe Algos.encode(nextBlock.header.stateRoot)
+
+      sourceAtSnapshot.closeStorage()
+    }
+  }
+
+  property("reject a prepared UTXO snapshot state from a noncanonical fork on restart") {
+    val protoSettings = NodeViewTestConfig(StateType.Utxo, verifyTransactions = true, popowBootstrap = false)
+      .toSettings
+    val snapshotSettings = protoSettings.copy(
+      nodeSettings = protoSettings.nodeSettings.copy(
+        utxoSettings = protoSettings.nodeSettings.utxoSettings.copy(utxoBootstrap = true)
+      )
+    )
+
+    new NodeViewFixture(snapshotSettings, parameters).apply { fixture =>
+      import fixture._
+
+      val (sourceState, boxHolder) = createUtxoState(settings)
+      val transactions = validTransactionsFromBoxHolder(boxHolder, new RandomWrapper)._1
+      val firstTimestamp = System.currentTimeMillis()
+      val firstBlock = validFullBlock(None, sourceState, transactions, Some(firstTimestamp))
+      val secondBlock = validFullBlock(None, sourceState, transactions, Some(firstTimestamp + 1))
+      firstBlock.id should not be secondBlock.id
+      java.util.Arrays.equals(firstBlock.header.stateRoot, secondBlock.header.stateRoot) shouldBe true
+
+      applyHeader(firstBlock.header).get
+      applyHeader(secondBlock.header).get
+      val canonicalHeader = getHistory.bestHeaderAtHeight(firstBlock.height).get
+      val noncanonicalBlock = Seq(firstBlock, secondBlock).find(_.id != canonicalHeader.id).get
+      java.util.Arrays.equals(noncanonicalBlock.header.stateRoot, canonicalHeader.stateRoot) shouldBe true
+      getHistory.onUtxoSnapshotApplied(noncanonicalBlock.height)
+      stopNodeViewHolder()
+
+      val stateDir = new File(s"${nodeViewDir.getAbsolutePath}/state")
+      fixture.deleteRecursive(stateDir)
+      stateDir.mkdirs() shouldBe true
+      val persistedGenesis = ErgoState
+        .generateGenesisUtxoState(stateDir, settings, Some(parameters))
+        ._1
+      val persistedForkState = persistedGenesis
+        .applyModifier(noncanonicalBlock, None)(_ => ())
+        .get
+      persistedForkState.version shouldBe idToVersion(noncanonicalBlock.id)
+      persistedForkState.version should not be idToVersion(canonicalHeader.id)
+      java.util.Arrays.equals(persistedForkState.rootDigest, canonicalHeader.stateRoot) shouldBe true
+      persistedForkState.closeStorage()
+
+      startNodeViewHolder()
+
+      getRootHash shouldBe Algos.encode(settings.chainSettings.genesisStateDigest)
+
+      sourceState.closeStorage()
+    }
+  }
+
+  property("require every prepared UTXO snapshot trust signal") {
+    val (state, boxHolder) = createUtxoState(settings)
+
+    try {
+      val header = validFullBlock(None, state, boxHolder).header
+      val matchingVersion = idToVersion(header.id)
+      val mismatchedVersion = idToVersion(Header.GenesisParentId)
+      val mismatchedRoot = header.stateRoot.clone()
+      mismatchedRoot(0) = (mismatchedRoot(0) ^ 1).toByte
+
+      val cases = Seq(
+        ("all signals match", true, true, true, matchingVersion, header.stateRoot, Some(header), true),
+        ("state is not UTXO", false, true, true, matchingVersion, header.stateRoot, Some(header), false),
+        ("UTXO bootstrap disabled", true, false, true, matchingVersion, header.stateRoot, Some(header), false),
+        ("snapshot marker absent", true, true, false, matchingVersion, header.stateRoot, Some(header), false),
+        ("canonical header absent", true, true, true, matchingVersion, header.stateRoot, None, false),
+        ("state version mismatch", true, true, true, mismatchedVersion, header.stateRoot, Some(header), false),
+        ("state root mismatch", true, true, true, matchingVersion, mismatchedRoot, Some(header), false)
+      )
+
+      cases.foreach { case (clue, stateIsUtxo, utxoBootstrap, snapshotApplied, stateVersion, stateRoot, headerOpt, expected) =>
+        withClue(clue) {
+          ErgoNodeViewHolder.isPreparedUtxoSnapshotState(
+            stateIsUtxo,
+            utxoBootstrap,
+            snapshotApplied,
+            stateVersion,
+            stateRoot,
+            headerOpt) shouldBe expected
+        }
+      }
+
+      var utxoBootstrapRead = false
+      var snapshotMarkerRead = false
+      var snapshotHeaderRead = false
+      ErgoNodeViewHolder.isPreparedUtxoSnapshotState(
+        stateIsUtxo = false,
+        utxoBootstrap = {
+          utxoBootstrapRead = true
+          true
+        },
+        snapshotApplied = {
+          snapshotMarkerRead = true
+          true
+        },
+        stateVersion = matchingVersion,
+        stateRoot = header.stateRoot,
+        snapshotHeaderOpt = {
+          snapshotHeaderRead = true
+          Some(header)
+        }) shouldBe false
+      utxoBootstrapRead shouldBe false
+      snapshotMarkerRead shouldBe false
+      snapshotHeaderRead shouldBe false
+
+      ErgoNodeViewHolder.isPreparedUtxoSnapshotState(
+        stateIsUtxo = true,
+        utxoBootstrap = false,
+        snapshotApplied = {
+          snapshotMarkerRead = true
+          true
+        },
+        stateVersion = matchingVersion,
+        stateRoot = header.stateRoot,
+        snapshotHeaderOpt = {
+          snapshotHeaderRead = true
+          Some(header)
+        }) shouldBe false
+      snapshotMarkerRead shouldBe false
+      snapshotHeaderRead shouldBe false
+
+      ErgoNodeViewHolder.isPreparedUtxoSnapshotState(
+        stateIsUtxo = true,
+        utxoBootstrap = true,
+        snapshotApplied = false,
+        stateVersion = matchingVersion,
+        stateRoot = header.stateRoot,
+        snapshotHeaderOpt = {
+          snapshotHeaderRead = true
+          Some(header)
+        }) shouldBe false
+      snapshotHeaderRead shouldBe false
+    } finally {
+      state.closeStorage()
+    }
+  }
+
   val genesisIdTestCases = List(t14, t15, t16, t17, t18, t19)
 
   def genesisIdConfig(expectedGenesisIdOpt: Option[ModifierId])(protoSettings: ErgoSettings): ErgoSettings = {
@@ -650,6 +875,14 @@ class ErgoNodeViewHolderSpec extends ErgoCorePropertyTest with NodeViewTestOps w
     }
   }
 
+  property("nipopow proof starts utxo snapshot bootstrap when utxoBootstrap enabled") {
+    t22.run(parameters, NodeViewTestConfig(StateType.Utxo, verifyTransactions = true, popowBootstrap = true, utxoBootstrap = true))
+  }
+
+  property("nipopow proof starts full blocks downloading when utxoBootstrap disabled") {
+    t22.run(parameters, NodeViewTestConfig(StateType.Utxo, verifyTransactions = true, popowBootstrap = true, utxoBootstrap = false))
+  }
+
   property("extractFailedTxId should extract failing transaction id from validation error shapes") {
     forAll(invalidErgoTransactionGen) { tx =>
       // transaction-level error tagged with the transaction id
diff --git a/src/test/scala/org/ergoplatform/utils/HistoryTestHelpers.scala b/src/test/scala/org/ergoplatform/utils/HistoryTestHelpers.scala
index fd3aebbfaf..45ff6f2879 100644
--- a/src/test/scala/org/ergoplatform/utils/HistoryTestHelpers.scala
+++ b/src/test/scala/org/ergoplatform/utils/HistoryTestHelpers.scala
@@ -45,12 +45,13 @@ object HistoryTestHelpers extends FileUtils {
                       epochLength: Int = 100000000,
                       useLastEpochs: Int = 10,
                       initialDiffOpt: Option[BigInt] = None,
-                      genesisIdOpt: Option[ModifierId] = None): ErgoHistory = {
+                      genesisIdOpt: Option[ModifierId] = None,
+                      utxoBootstrap: Boolean = false): ErgoHistory = {
 
     val txCostLimit     = initSettings.nodeSettings.maxTransactionCost
     val txSizeLimit     = initSettings.nodeSettings.maxTransactionSize
     val nodeSettings: NodeConfigurationSettings = NodeConfigurationSettings(stateType, verifyTransactions, blocksToKeep,
-      UtxoSettings(false, 0, 2), NipopowSettings(false, 1), mining = false, txCostLimit, txSizeLimit, blockCandidateGenerationInterval = 20.seconds,
+      UtxoSettings(utxoBootstrap, 0, 2), NipopowSettings(false, 1), mining = false, txCostLimit, txSizeLimit, blockCandidateGenerationInterval = 20.seconds,
       useExternalMiner = false, internalMinersCount = 1, internalMinerPollingInterval = 1.second, miningPubKeyHex = None,
       offlineGeneration = false, 200, 5.minutes, 100000, 1.minute, mempoolSorting = SortingOption.FeePerByte,
       rebroadcastCount = 200, 1000000, 100, adProofsSuffixLength = 112*1024, extraIndex = false
diff --git a/src/test/scala/org/ergoplatform/utils/NodeViewTestConfig.scala b/src/test/scala/org/ergoplatform/utils/NodeViewTestConfig.scala
index 3b708d9d06..f2ef2e6e0c 100644
--- a/src/test/scala/org/ergoplatform/utils/NodeViewTestConfig.scala
+++ b/src/test/scala/org/ergoplatform/utils/NodeViewTestConfig.scala
@@ -7,7 +7,8 @@ import org.ergoplatform.settings.{ErgoSettings, ErgoSettingsReader, NipopowSetti
 
 case class NodeViewTestConfig(stateType: StateType,
                               verifyTransactions: Boolean,
-                              popowBootstrap: Boolean) {
+                              popowBootstrap: Boolean,
+                              utxoBootstrap: Boolean = false) {
 
   def toSettings: ErgoSettings = {
     val defaultSettings = ErgoSettingsReader.read()
@@ -18,13 +19,15 @@ case class NodeViewTestConfig(stateType: StateType,
       nodeSettings = defaultSettings.nodeSettings.copy(
         stateType = stateType,
         verifyTransactions = verifyTransactions,
-        nipopowSettings = NipopowSettings(popowBootstrap, 1)
+        nipopowSettings = NipopowSettings(popowBootstrap, 1),
+        utxoSettings = defaultSettings.nodeSettings.utxoSettings.copy(utxoBootstrap = utxoBootstrap)
       )
     )
   }
 
   override def toString: String = {
-    s"State: $stateType, Verify Transactions: $verifyTransactions, PoPoW Bootstrap: $popowBootstrap"
+    s"State: $stateType, Verify Transactions: $verifyTransactions, PoPoW Bootstrap: $popowBootstrap, " +
+      s"UTXO Bootstrap: $utxoBootstrap"
   }
 }