mirrors/bitcoin-s
1
0
Fork 0
mirror of https://github.com/bitcoin-s/bitcoin-s.git synced 2024-11-19 01:40:55 +01:00
Code Issues Packages Projects Releases Wiki Activity

Persist whether wallet is rescanning in the database (#4326)

Browse Source 
* Persist whether wallet is rescanning in the database

* fix cli

* fix build

* fix unit tests

* fix postgres tests

* remove wallet_state table

* fix rescan bug

* cleanup

* revert Cancellable's

* Cleanup

Co-authored-by: Chris Stewart <stewart.chris1234@gmail.com>
This commit is contained in:
rorp 2022-05-23 17:03:02 -07:00 committed by GitHub
parent 527e3ae862
commit f680ab8691
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
13 changed files with 349 additions and 192 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
app-commons/src/main/scala/org/bitcoins/commons/jsonmodels/wallet/WalletStateDescriptor.scala
View File

@ -9,7 +9,9 @@ object WalletStateDescriptorType
final case object SyncHeight extends WalletStateDescriptorType
val all: Vector[WalletStateDescriptorType] = Vector(SyncHeight)
final case object Rescan extends WalletStateDescriptorType
val all: Vector[WalletStateDescriptorType] = Vector(SyncHeight, Rescan)
override def fromStringOpt(str: String): Option[WalletStateDescriptorType] = {
all.find(state => str.toLowerCase() == state.toString.toLowerCase)
@ -36,8 +38,8 @@ sealed trait WalletStateDescriptorFactory[T <: WalletStateDescriptor]
object WalletStateDescriptor extends StringFactory[WalletStateDescriptor] {
val all: Vector[StringFactory[WalletStateDescriptor]] = Vector(
SyncHeightDescriptor)
val all: Vector[StringFactory[WalletStateDescriptor]] =
Vector(SyncHeightDescriptor, RescanDescriptor)
override def fromString(string: String): WalletStateDescriptor = {
all.find(f => f.fromStringT(string).isSuccess) match {
@ -71,3 +73,22 @@ object SyncHeightDescriptor
SyncHeightDescriptor(hash, height)
}
}
case class RescanDescriptor(rescanning: Boolean) extends WalletStateDescriptor {
override val descriptorType: WalletStateDescriptorType =
WalletStateDescriptorType.Rescan
override val toString: String = rescanning.toString
}
object RescanDescriptor extends WalletStateDescriptorFactory[RescanDescriptor] {
override val tpe: WalletStateDescriptorType =
WalletStateDescriptorType.Rescan
override def fromString(string: String): RescanDescriptor = {
val rescanning = java.lang.Boolean.parseBoolean(string)
RescanDescriptor(rescanning)
}
}

12
app/server-test/src/test/scala/org/bitcoins/server/RoutesSpec.scala
View File

@ -1697,8 +1697,9 @@ class RoutesSpec extends AnyWordSpec with ScalatestRouteTest with MockFactory {
.rescanNeutrinoWallet(_: Option[BlockStamp],
_: Option[BlockStamp],
_: Int,
_: Boolean,
_: Boolean)(_: ExecutionContext))
.expects(None, None, 100, false, executor)
.expects(None, None, 100, false, false, executor)
.returning(Future.successful(RescanState.RescanDone))
val route1 =
@ -1718,6 +1719,7 @@ class RoutesSpec extends AnyWordSpec with ScalatestRouteTest with MockFactory {
.rescanNeutrinoWallet(_: Option[BlockStamp],
_: Option[BlockStamp],
_: Int,
_: Boolean,
_: Boolean)(_: ExecutionContext))
.expects(
Some(BlockTime(
@ -1725,6 +1727,7 @@ class RoutesSpec extends AnyWordSpec with ScalatestRouteTest with MockFactory {
None,
100,
false,
false,
executor)
.returning(Future.successful(RescanState.RescanDone))
@ -1747,11 +1750,13 @@ class RoutesSpec extends AnyWordSpec with ScalatestRouteTest with MockFactory {
.rescanNeutrinoWallet(_: Option[BlockStamp],
_: Option[BlockStamp],
_: Int,
_: Boolean,
_: Boolean)(_: ExecutionContext))
.expects(None,
Some(BlockHash(DoubleSha256DigestBE.empty)),
100,
false,
false,
executor)
.returning(Future.successful(RescanState.RescanDone))
@ -1774,11 +1779,13 @@ class RoutesSpec extends AnyWordSpec with ScalatestRouteTest with MockFactory {
.rescanNeutrinoWallet(_: Option[BlockStamp],
_: Option[BlockStamp],
_: Int,
_: Boolean,
_: Boolean)(_: ExecutionContext))
.expects(Some(BlockHeight(12345)),
Some(BlockHeight(67890)),
100,
false,
false,
executor)
.returning(Future.successful(RescanState.RescanDone))
@ -1838,8 +1845,9 @@ class RoutesSpec extends AnyWordSpec with ScalatestRouteTest with MockFactory {
.rescanNeutrinoWallet(_: Option[BlockStamp],
_: Option[BlockStamp],
_: Int,
_: Boolean,
_: Boolean)(_: ExecutionContext))
.expects(None, None, 55, false, executor)
.expects(None, None, 55, false, false, executor)
.returning(Future.successful(RescanState.RescanDone))
val route8 =

25
app/server/src/main/scala/org/bitcoins/server/BitcoinSServerMain.scala
View File

@ -211,6 +211,7 @@ class BitcoinSServerMain(override val serverArgParser: ServerArgParser)(implicit
_ <- startedTorConfigF
wallet <- configuredWalletF
_ <- handleDuplicateSpendingInfoDb(wallet)
_ <- restartRescanIfNeeded(wallet)
_ <- node.sync()
} yield {
logger.info(
@ -337,6 +338,7 @@ class BitcoinSServerMain(override val serverArgParser: ServerArgParser)(implicit
_ = dlcConf.addCallbacks(dlcWalletCallbacks)
_ <- startedTorConfigF
_ <- handleDuplicateSpendingInfoDb(wallet)
_ <- restartRescanIfNeeded(wallet)
} yield {
logger.info(s"Done starting Main!")
()
@ -480,7 +482,8 @@ class BitcoinSServerMain(override val serverArgParser: ServerArgParser)(implicit
endOpt = None,
addressBatchSize =
walletConf.discoveryBatchSize,
useCreationTime = true)
useCreationTime = true,
force = true)
} yield clearedWallet
walletF.map(_ => ())
}
@ -506,7 +509,7 @@ class BitcoinSServerMain(override val serverArgParser: ServerArgParser)(implicit
.startBitcoindBlockPolling(wallet, bitcoind)
.map { _ =>
BitcoindRpcBackendUtil
.startBitcoindMempoolPolling(bitcoind) { tx =>
.startBitcoindMempoolPolling(wallet, bitcoind) { tx =>
nodeConf.nodeCallbacks
.executeOnTxReceivedCallbacks(logger, tx)
}
@ -565,7 +568,8 @@ class BitcoinSServerMain(override val serverArgParser: ServerArgParser)(implicit
.rescanNeutrinoWallet(startOpt = None,
endOpt = None,
addressBatchSize = wallet.discoveryBatchSize,
useCreationTime = true)
useCreationTime = true,
force = true)
.recover { case scala.util.control.NonFatal(exn) =>
logger.error(s"Failed to handleDuplicateSpendingInfoDb rescan",
exn)
@ -577,6 +581,21 @@ class BitcoinSServerMain(override val serverArgParser: ServerArgParser)(implicit
_ <- spendingInfoDAO.createOutPointsIndexIfNeeded()
} yield ()
}
private def restartRescanIfNeeded(wallet: Wallet): Future[RescanState] = {
for {
isRescanning <- wallet.isRescanning()
res <-
if (isRescanning)
wallet.rescanNeutrinoWallet(startOpt = None,
endOpt = None,
addressBatchSize =
wallet.discoveryBatchSize,
useCreationTime = true,
force = true)
else Future.successful(RescanState.RescanDone)
} yield res
}
}
object BitcoinSServerMain extends BitcoinSAppScalaDaemon {

228
app/server/src/main/scala/org/bitcoins/server/BitcoindRpcBackendUtil.scala
View File

@ -291,8 +291,8 @@ object BitcoindRpcBackendUtil extends Logging {
* if it has changed, it will then request those blocks to process them
*
* @param startCount The starting block height of the wallet
* @param interval The amount of time between polls, this should not be too aggressive
* as the wallet will need to process the new blocks
* @param interval The amount of time between polls, this should not be too aggressive
* as the wallet will need to process the new blocks
*/
def startBitcoindBlockPolling(
wallet: WalletApi,
@ -300,73 +300,91 @@ object BitcoindRpcBackendUtil extends Logging {
interval: FiniteDuration = 10.seconds)(implicit
system: ActorSystem,
ec: ExecutionContext): Future[Cancellable] = {
val walletSyncStateF = wallet.getSyncState()
val resultF: Future[Cancellable] = for {
walletSyncState <- walletSyncStateF
for {
walletSyncState <- wallet.getSyncState()
} yield {
val numParallelism = Runtime.getRuntime.availableProcessors()
val atomicPrevCount: AtomicInteger = new AtomicInteger(
walletSyncState.height)
val processing = new AtomicBoolean(false)
val processingBitcoindBlocks = new AtomicBoolean(false)
def pollBitcoind(): Future[Unit] = {
if (processingBitcoindBlocks.compareAndSet(false, true)) {
logger.trace("Polling bitcoind for block count")
val res: Future[Unit] = bitcoind.getBlockCount.flatMap { count =>
val prevCount = atomicPrevCount.get()
if (prevCount < count) {
logger.info(
s"Bitcoind has new block(s), requesting... ${count - prevCount} blocks")
// use .tail so we don't process the previous block that we already did
val range = prevCount.to(count).tail
val hashFs: Future[Seq[DoubleSha256Digest]] = Source(range)
.mapAsync(parallelism = numParallelism) { height =>
bitcoind.getBlockHash(height).map(_.flip)
}
.map { hash =>
val _ = atomicPrevCount.incrementAndGet()
hash
}
.toMat(Sink.seq)(Keep.right)
.run()
val requestsBlocksF = for {
hashes <- hashFs
_ <- wallet.nodeApi.downloadBlocks(hashes.toVector)
} yield logger.debug(
"Successfully polled bitcoind for new blocks")
requestsBlocksF.failed.foreach { case err =>
val failedCount = atomicPrevCount.get
atomicPrevCount.set(prevCount)
logger.error(
s"Requesting blocks from bitcoind polling failed, range=[$prevCount, $failedCount]",
err)
}
requestsBlocksF
} else if (prevCount > count) {
Future.failed(new RuntimeException(
s"Bitcoind is at a block height ($count) before the wallet's ($prevCount)"))
} else {
logger.debug(s"In sync $prevCount count=$count")
Future.unit
}
}
res.onComplete(_ => processingBitcoindBlocks.set(false))
res
} else {
logger.info(
s"Skipping scanning the blockchain since a previously scheduled task is still running")
Future.unit
}
}
system.scheduler.scheduleWithFixedDelay(0.seconds, interval) { () =>
{
if (processing.compareAndSet(false, true)) {
logger.trace("Polling bitcoind for block count")
val res = bitcoind.getBlockCount.flatMap { count =>
val prevCount = atomicPrevCount.get()
if (prevCount < count) {
logger.info(
s"Bitcoind has new block(s), requesting... ${count - prevCount} blocks")
// use .tail so we don't process the previous block that we already did
val range = prevCount.to(count).tail
val hashFs: Future[Seq[DoubleSha256Digest]] = Source(range)
.mapAsync(parallelism = numParallelism) { height =>
bitcoind.getBlockHash(height).map(_.flip)
}
.map { hash =>
val _ = atomicPrevCount.incrementAndGet()
hash
}
.toMat(Sink.seq)(Keep.right)
.run()
val requestsBlocksF = for {
hashes <- hashFs
_ <- wallet.nodeApi.downloadBlocks(hashes.toVector)
} yield logger.debug(
"Successfully polled bitcoind for new blocks")
requestsBlocksF.failed.foreach { case err =>
val failedCount = atomicPrevCount.get
atomicPrevCount.set(prevCount)
logger.error(
s"Requesting blocks from bitcoind polling failed, range=[$prevCount, $failedCount]",
err)
}
requestsBlocksF
} else if (prevCount > count) {
Future.failed(new RuntimeException(
s"Bitcoind is at a block height ($count) before the wallet's ($prevCount)"))
val f = for {
rescanning <- wallet.isRescanning()
res <-
if (!rescanning) {
pollBitcoind()
} else {
logger.debug(s"In sync $prevCount count=$count")
logger.info(
s"Skipping scanning the blockchain during wallet rescan")
Future.unit
}
}
res.onComplete(_ => processing.set(false))
} else {
logger.info(
s"Skipping scanning the blockchain since a previously scheduled task is still running")
}
} yield res
f.failed.foreach(err => logger.error(s"Failed to poll bitcoind", err))
}
}
}
resultF
}
def startBitcoindMempoolPolling(
wallet: WalletApi,
bitcoind: BitcoindRpcClient,
interval: FiniteDuration = 10.seconds)(
processTx: Transaction => Future[Unit])(implicit
@ -383,52 +401,70 @@ object BitcoindRpcBackendUtil extends Logging {
txids
}
val processing = new AtomicBoolean(false)
val processingMempool = new AtomicBoolean(false)
def pollMempool(): Future[Unit] = {
if (processingMempool.compareAndSet(false, true)) {
logger.debug("Polling bitcoind for mempool")
val numParallelism = {
val processors = Runtime.getRuntime.availableProcessors()
//max open requests is 32 in akka, so 1/8 of possible requests
//can be used to query the mempool, else just limit it be number of processors
//see: https://github.com/bitcoin-s/bitcoin-s/issues/4252
Math.min(4, processors)
}
//don't want to execute these in parallel
val processTxFlow = Sink.foreachAsync[Transaction](1)(processTx)
val res = for {
mempool <- bitcoind.getRawMemPool
newTxIds = getDiffAndReplace(mempool.toSet)
_ = logger.debug(s"Found ${newTxIds.size} new mempool transactions")
_ <- Source(newTxIds)
.mapAsync(parallelism = numParallelism) { txid =>
bitcoind
.getRawTransactionRaw(txid)
.map(Option(_))
.recover { case _: Throwable =>
None
}
.collect { case Some(tx) =>
tx
}
}
.toMat(processTxFlow)(Keep.right)
.run()
} yield {
logger.debug(
s"Done processing ${newTxIds.size} new mempool transactions")
()
}
res.onComplete(_ => processingMempool.set(false))
res
} else {
logger.info(
s"Skipping scanning the mempool since a previously scheduled task is still running")
Future.unit
}
}
system.scheduler.scheduleWithFixedDelay(0.seconds, interval) { () =>
{
if (processing.compareAndSet(false, true)) {
logger.debug("Polling bitcoind for mempool")
val numParallelism = {
val processors = Runtime.getRuntime.availableProcessors()
//max open requests is 32 in akka, so 1/8 of possible requests
//can be used to query the mempool, else just limit it be number of processors
//see: https://github.com/bitcoin-s/bitcoin-s/issues/4252
Math.min(4, processors)
}
val f = for {
rescanning <- wallet.isRescanning()
res <-
if (!rescanning) {
pollMempool()
} else {
logger.info(s"Skipping scanning the mempool during wallet rescan")
Future.unit
}
} yield res
//don't want to execute these in parallel
val processTxFlow = Sink.foreachAsync[Transaction](1)(processTx)
val res = for {
mempool <- bitcoind.getRawMemPool
newTxIds = getDiffAndReplace(mempool.toSet)
_ = logger.debug(s"Found ${newTxIds.size} new mempool transactions")
_ <- Source(newTxIds)
.mapAsync(parallelism = numParallelism) { txid =>
bitcoind
.getRawTransactionRaw(txid)
.map(Option(_))
.recover { case _: Throwable =>
None
}
.collect { case Some(tx) =>
tx
}
}
.toMat(processTxFlow)(Keep.right)
.run()
} yield {
logger.debug(
s"Done processing ${newTxIds.size} new mempool transactions")
()
}
res.onComplete(_ => processing.set(false))
} else {
logger.info(
s"Skipping scanning the mempool since a previously scheduled task is still running")
}
f.failed.foreach(err => logger.error(s"Failed to poll mempool", err))
()
}
}
}

3
app/server/src/main/scala/org/bitcoins/server/WalletRoutes.scala
View File

@ -711,7 +711,8 @@ case class WalletRoutes(wallet: AnyDLCHDWalletApi)(implicit
endOpt = endBlock,
addressBatchSize =
batchSize.getOrElse(wallet.discoveryBatchSize()),
useCreationTime = !ignoreCreationTime)
useCreationTime = !ignoreCreationTime,
force = false)
Future.successful("Rescan started.")
} else {
Future.successful(

11
core/src/main/scala/org/bitcoins/core/api/wallet/NeutrinoWalletApi.scala
View File

@ -76,16 +76,17 @@ trait NeutrinoWalletApi { self: WalletApi =>
startOpt: Option[BlockStamp],
endOpt: Option[BlockStamp],
addressBatchSize: Int,
useCreationTime: Boolean)(implicit
ec: ExecutionContext): Future[RescanState]
useCreationTime: Boolean,
force: Boolean)(implicit ec: ExecutionContext): Future[RescanState]
/** Helper method to rescan the ENTIRE blockchain. */
def fullRescanNeutrinoWallet(addressBatchSize: Int)(implicit
ec: ExecutionContext): Future[RescanState] =
def fullRescanNeutrinoWallet(addressBatchSize: Int, force: Boolean = false)(
implicit ec: ExecutionContext): Future[RescanState] =
rescanNeutrinoWallet(startOpt = None,
endOpt = None,
addressBatchSize = addressBatchSize,
useCreationTime = false)
useCreationTime = false,
force = force)
def discoveryBatchSize(): Int

6
dlc-wallet-test/src/test/scala/org/bitcoins/dlc/wallet/RescanDLCTest.scala
View File

@ -56,7 +56,8 @@ class RescanDLCTest extends DualWalletTestCachedBitcoind {
_ <- wallet.rescanNeutrinoWallet(startOpt = None,
endOpt = Some(BlockHash(hash)),
addressBatchSize = 20,
useCreationTime = false)
useCreationTime = false,
force = false)
postStatus <- getDLCStatus(wallet)
} yield assert(postStatus.state == DLCState.Claimed)
@ -96,7 +97,8 @@ class RescanDLCTest extends DualWalletTestCachedBitcoind {
_ <- wallet.rescanNeutrinoWallet(startOpt = None,
endOpt = Some(BlockHash(hash)),
addressBatchSize = 20,
useCreationTime = false)
useCreationTime = false,
force = false)
postStatus <- getDLCStatus(wallet)
} yield assert(postStatus.state == DLCState.RemoteClaimed)

3
wallet-test/src/test/scala/org/bitcoins/wallet/BitcoindBlockPollingTest.scala
View File

@ -82,7 +82,8 @@ class BitcoindBlockPollingTest
txid1 <- bitcoind.sendToAddress(addr, amountToSend)
// Setup block polling
_ = BitcoindRpcBackendUtil.startBitcoindMempoolPolling(bitcoind,
_ = BitcoindRpcBackendUtil.startBitcoindMempoolPolling(wallet,
bitcoind,
1.second) { tx =>
mempoolTxs += tx
FutureUtil.unit

24
wallet-test/src/test/scala/org/bitcoins/wallet/RescanHandlingTest.scala
View File

@ -141,7 +141,8 @@ class RescanHandlingTest extends BitcoinSWalletTestCachedBitcoindNewest {
endOpt = None,
addressBatchSize =
DEFAULT_ADDR_BATCH_SIZE,
useCreationTime = false)
useCreationTime = false,
force = false)
balance <- newTxWallet.getBalance()
unconfirmedBalance <- newTxWallet.getUnconfirmedBalance()
} yield {
@ -250,7 +251,8 @@ class RescanHandlingTest extends BitcoinSWalletTestCachedBitcoindNewest {
endOpt = None,
addressBatchSize =
DEFAULT_ADDR_BATCH_SIZE,
useCreationTime = true)
useCreationTime = true,
force = false)
balance <- newTxWallet.getBalance()
unconfirmedBalance <- newTxWallet.getUnconfirmedBalance()
} yield {
@ -289,7 +291,8 @@ class RescanHandlingTest extends BitcoinSWalletTestCachedBitcoindNewest {
endOpt = end,
addressBatchSize =
DEFAULT_ADDR_BATCH_SIZE,
useCreationTime = false)
useCreationTime = false,
force = false)
balanceAfterRescan <- wallet.getBalance()
} yield {
assert(balanceAfterRescan == CurrencyUnits.zero)
@ -306,7 +309,8 @@ class RescanHandlingTest extends BitcoinSWalletTestCachedBitcoindNewest {
endOpt = None,
addressBatchSize =
DEFAULT_ADDR_BATCH_SIZE,
useCreationTime = true)
useCreationTime = true,
force = false)
//slight delay to make sure other rescan is started
val alreadyStartedF =
@ -315,7 +319,8 @@ class RescanHandlingTest extends BitcoinSWalletTestCachedBitcoindNewest {
endOpt = None,
addressBatchSize =
DEFAULT_ADDR_BATCH_SIZE,
useCreationTime = true)
useCreationTime = true,
force = false)
}
for {
start <- startF
@ -338,7 +343,8 @@ class RescanHandlingTest extends BitcoinSWalletTestCachedBitcoindNewest {
_ <- wallet.rescanNeutrinoWallet(startOpt = None,
endOpt = None,
addressBatchSize = 10,
useCreationTime = true)
useCreationTime = true,
force = false)
usedAddresses <- wallet.listFundedAddresses()
@ -370,13 +376,15 @@ class RescanHandlingTest extends BitcoinSWalletTestCachedBitcoindNewest {
_ <- wallet.rescanNeutrinoWallet(startOpt = None,
endOpt = None,
addressBatchSize = 10,
useCreationTime = true)
useCreationTime = true,
force = false)
addressNoFunds <- wallet.getNewChangeAddress()
//rescan again
_ <- wallet.rescanNeutrinoWallet(startOpt = None,
endOpt = None,
addressBatchSize = 10,
useCreationTime = true)
useCreationTime = true,
force = false)
txid <- bitcoind.sendToAddress(addressNoFunds, amt)
tx <- bitcoind.getRawTransactionRaw(txid)
_ <- wallet.processTransaction(tx, None)

2
wallet/src/main/scala/org/bitcoins/wallet/Wallet.scala
View File

@ -159,7 +159,7 @@ abstract class Wallet
override def stop(): Future[Wallet] = Future.successful(this)
def getSyncDescriptorOpt(): Future[Option[SyncHeightDescriptor]] = {
stateDescriptorDAO.getSyncDescriptorOpt()
stateDescriptorDAO.getSyncHeight()
}
override def getSyncState(): Future[BlockSyncState] = {

22
wallet/src/main/scala/org/bitcoins/wallet/db/WalletDbManagement.scala
View File

@ -57,16 +57,18 @@ trait WalletDbManagement extends DbManagement {
// Ordering matters here, tables with a foreign key should be listed after
// the table that key references
override lazy val allTables: List[TableQuery[Table[_]]] = {
List(spkTable,
accountTable,
addressTable,
addressTagTable,
txTable,
incomingTxTable,
utxoTable,
outgoingTxTable,
stateDescriptorTable,
masterXPubTable)
List(
spkTable,
accountTable,
addressTable,
addressTagTable,
txTable,
incomingTxTable,
utxoTable,
outgoingTxTable,
stateDescriptorTable,
masterXPubTable
)
}
}

100
wallet/src/main/scala/org/bitcoins/wallet/internal/RescanHandling.scala
View File

@ -15,35 +15,31 @@ import org.bitcoins.core.wallet.rescan.RescanState
import org.bitcoins.crypto.DoubleSha256Digest
import org.bitcoins.wallet.{Wallet, WalletLogger}
import java.util.concurrent.atomic.AtomicBoolean
import scala.concurrent.{ExecutionContext, Future}
import scala.util.{Failure, Success}
private[wallet] trait RescanHandling extends WalletLogger {
self: Wallet =>
private val rescanning = new AtomicBoolean(false)
/////////////////////
// Public facing API
override def isRescanning(): Future[Boolean] =
Future.successful(rescanning.get())
override def isRescanning(): Future[Boolean] = stateDescriptorDAO.isRescanning
/** @inheritdoc */
override def rescanNeutrinoWallet(
startOpt: Option[BlockStamp],
endOpt: Option[BlockStamp],
addressBatchSize: Int,
useCreationTime: Boolean)(implicit
ec: ExecutionContext): Future[RescanState] = {
useCreationTime: Boolean,
force: Boolean)(implicit ec: ExecutionContext): Future[RescanState] = {
for {
account <- getDefaultAccount()
state <- rescanNeutrinoWallet(account.hdAccount,
startOpt,
endOpt,
addressBatchSize,
useCreationTime)
useCreationTime,
force)
} yield state
}
@ -53,45 +49,55 @@ private[wallet] trait RescanHandling extends WalletLogger {
startOpt: Option[BlockStamp],
endOpt: Option[BlockStamp],
addressBatchSize: Int,
useCreationTime: Boolean = true): Future[RescanState] = {
if (rescanning.get()) {
logger.warn(
s"Rescan already started, ignoring request to start another one")
Future.successful(RescanState.RescanInProgress)
} else {
rescanning.set(true)
logger.info(
s"Starting rescanning the wallet from ${startOpt} to ${endOpt} useCreationTime=$useCreationTime")
val start = System.currentTimeMillis()
val res = for {
start <- (startOpt, useCreationTime) match {
case (Some(_), true) =>
Future.failed(new IllegalArgumentException(
"Cannot define a starting block and use the wallet creation time"))
case (Some(value), false) =>
Future.successful(Some(value))
case (None, true) =>
walletCreationBlockHeight.map(Some(_))
case (None, false) =>
Future.successful(None)
}
_ <- clearUtxos(account)
_ <- doNeutrinoRescan(account, start, endOpt, addressBatchSize)
} yield {
RescanState.RescanDone
}
res.onComplete {
case Success(_) =>
rescanning.set(false)
useCreationTime: Boolean = true,
force: Boolean = false): Future[RescanState] = {
for {
doRescan <-
if (force) stateDescriptorDAO.updateRescanning(true).map(_.rescanning)
else
stateDescriptorDAO.compareAndSetRescanning(expectedValue = false,
newValue = true)
rescanState <-
if (doRescan) {
logger.info(
s"Finished rescanning the wallet. It took ${System.currentTimeMillis() - start}ms")
case Failure(err) =>
rescanning.set(false)
logger.error(s"Failed to rescan wallet", err)
}
res
}
s"Starting rescanning the wallet from ${startOpt} to ${endOpt} useCreationTime=$useCreationTime")
val startTime = System.currentTimeMillis()
val res = for {
start <- (startOpt, useCreationTime) match {
case (Some(_), true) =>
Future.failed(new IllegalArgumentException(
"Cannot define a starting block and use the wallet creation time"))
case (Some(value), false) =>
Future.successful(Some(value))
case (None, true) =>
walletCreationBlockHeight.map(Some(_))
case (None, false) =>
Future.successful(None)
}
_ <- clearUtxos(account)
_ <- doNeutrinoRescan(account, start, endOpt, addressBatchSize)
_ <- stateDescriptorDAO.updateRescanning(false)
} yield {
logger.info(s"Finished rescanning the wallet. It took ${System
.currentTimeMillis() - startTime}ms")
RescanState.RescanDone
}
res.recoverWith { case err: Throwable =>
logger.error(s"Failed to rescan wallet", err)
stateDescriptorDAO
.updateRescanning(false)
.flatMap(_ => Future.failed(err))
}
res
} else {
logger.warn(
s"Rescan already started, ignoring request to start another one")
Future.successful(RescanState.RescanInProgress)
}
} yield rescanState
}
/** @inheritdoc */

78
wallet/src/main/scala/org/bitcoins/wallet/models/WalletStateDescriptorDAO.scala
View File

@ -2,6 +2,7 @@ package org.bitcoins.wallet.models
import org.bitcoins.commons.jsonmodels.wallet.WalletStateDescriptorType._
import org.bitcoins.commons.jsonmodels.wallet.{
RescanDescriptor,
SyncHeightDescriptor,
WalletStateDescriptor,
WalletStateDescriptorType
@ -55,7 +56,7 @@ case class WalletStateDescriptorDAO()(implicit
Seq] =
findByPrimaryKeys(ts.map(_.tpe))
def getSyncDescriptorOpt(): Future[Option[SyncHeightDescriptor]] = {
def getSyncHeight(): Future[Option[SyncHeightDescriptor]] = {
read(SyncHeight).map {
case Some(db) =>
val desc = SyncHeightDescriptor.fromString(db.descriptor.toString)
@ -67,22 +68,73 @@ case class WalletStateDescriptorDAO()(implicit
def updateSyncHeight(
hash: DoubleSha256DigestBE,
height: Int): Future[WalletStateDescriptorDb] = {
getSyncDescriptorOpt().flatMap {
case Some(old) =>
if (old.height > height) {
Future.successful(WalletStateDescriptorDb(SyncHeight, old))
} else {
val tpe: WalletStateDescriptorType = SyncHeight
val query = table.filter(_.tpe === tpe)
val action = for {
oldOpt <- query.result.headOption
res: WalletStateDescriptorDb <- oldOpt match {
case Some(oldDb) =>
val old = SyncHeightDescriptor.fromString(oldDb.descriptor.toString)
if (old.height > height) {
DBIO.successful(WalletStateDescriptorDb(tpe, old))
} else {
val descriptor = SyncHeightDescriptor(hash, height)
val newDb = WalletStateDescriptorDb(tpe, descriptor)
query.update(newDb).map(_ => newDb)
}
case None =>
val descriptor = SyncHeightDescriptor(hash, height)
val newDb = WalletStateDescriptorDb(SyncHeight, descriptor)
update(newDb)
}
case None =>
val descriptor = SyncHeightDescriptor(hash, height)
val db = WalletStateDescriptorDb(SyncHeight, descriptor)
create(db)
val db = WalletStateDescriptorDb(tpe, descriptor)
(table += db).map(_ => db)
}
} yield res
safeDatabase.run(action)
}
def getRescan(): Future[Option[RescanDescriptor]] = {
read(Rescan).map {
case Some(db) =>
val desc = RescanDescriptor.fromString(db.descriptor.toString)
Some(desc)
case None => None
}
}
def isRescanning: Future[Boolean] = getRescan().map(_.exists(_.rescanning))
def updateRescanning(rescanning: Boolean): Future[RescanDescriptor] = {
val desc = RescanDescriptor(rescanning)
upsert(WalletStateDescriptorDb(desc.descriptorType, desc)).map(_ => desc)
}
def compareAndSetRescanning(
expectedValue: Boolean,
newValue: Boolean): Future[Boolean] = {
val tpe: WalletStateDescriptorType = Rescan
val query = table.filter(_.tpe === tpe)
val actions = for {
dbs <- query.result
res <- dbs.headOption match {
case None =>
val desc = RescanDescriptor(newValue)
val db = WalletStateDescriptorDb(tpe, desc)
(table += db).map(_ => true)
case Some(db) =>
val oldDesc = RescanDescriptor.fromString(db.descriptor.toString)
if (oldDesc.rescanning == expectedValue) {
val newDesc = RescanDescriptor(true)
val newDb = WalletStateDescriptorDb(tpe, newDesc)
query.update(newDb).map(_ => true)
} else {
DBIO.successful(false)
}
}
} yield res
safeDatabase.run(actions)
}
class WalletStateDescriptorTable(t: Tag)
extends Table[WalletStateDescriptorDb](t,
schemaName,