2020 03 06 wallet rescan test (#1218)

* Add wallet rescan tests

* Create FilterSync, which gives us an API inside of the chain project to sync filters with

Add another unit test to filter sync

Add more unit tests for ChainSync and FilterSync

Clean up some docs, remove some extra lines of code

Run scalafmt

Add filter-sync.md

Cleanup some nits

Add more information of how FilterSync.syncFilters() works

Add 'FilterWithHeaderHash' type so that we can actually validate/verify block headers that are being fed into the chain project

Run scalafmt, hide imports in filter-sync.md so code appears cleaner

Move implicits out of invisible block as it seems to cause errors

Make it so FilterSync processes filters in batches rather than fetching them all at once

Fix compile error

* WIP bitcoind implement ChainQueryApi

* rework fixtures to be able to support injecting ChainQueryApi implemented by bitcoind into our fixture infrastructure for creating wallets

* Fix rebase problem

* Implement getFiltersBetweenHeight() with ben's solution

* WIP Start implementing NodeApi against bitcoind

* Actually inject the bitcoind backed nodeApi into our fixture

* Get first rescan test working for rescanning the entire blockchain

* Implement test case for rescanning from a specific height

* Fix NeutrinoNodeWalletTest test case that uses a experimental version of bitcoind, for now i believe our experimental binary is on v18

* Add wallet-rescan.md, add helper method to WalletApi that allows you to clear out the wallet's utxos/addresses

* Add another log to try to debug CI

* Address code review

* Fix wrong ordering of deletion of tables in clearUtxosAndAddresses()

* reset logging level

bitcoind-rpc/src/main/scala/org/bitcoins/rpc/client/common/BitcoindRpcClient.scala

@@ -94,7 +94,9 @@ object BitcoindRpcClient {
       case BitcoindVersion.V17 => BitcoindV17RpcClient.withActorSystem(instance)
       case BitcoindVersion.V18 => BitcoindV18RpcClient.withActorSystem(instance)
       case BitcoindVersion.V19 => BitcoindV19RpcClient.withActorSystem(instance)
-      case BitcoindVersion.Experimental | BitcoindVersion.Unknown =>
+      case BitcoindVersion.Experimental =>
+        BitcoindV18RpcClient.withActorSystem(instance)
+      case BitcoindVersion.Unknown =>
         sys.error(
           s"Cannot create a bitcoind from a unknown or experimental version")
     }

chain/src/main/scala/org/bitcoins/chain/blockchain/sync/FilterWithHeaderHash.scala

@@ -1,6 +1,6 @@
 package org.bitcoins.chain.blockchain.sync
 
-import org.bitcoins.core.crypto.{DoubleSha256Digest, DoubleSha256DigestBE}
+import org.bitcoins.core.crypto.DoubleSha256DigestBE
 import org.bitcoins.core.gcs.GolombFilter
 
 /** Represents a [[GolombFilter]] with it's [[org.bitcoins.core.gcs.FilterHeader]] associated with it

core/src/main/scala/org/bitcoins/core/api/ChainQueryApi.scala

@@ -4,7 +4,7 @@ import org.bitcoins.core.crypto.DoubleSha256DigestBE
 import org.bitcoins.core.gcs.GolombFilter
 import org.bitcoins.core.protocol.BlockStamp
 
-import scala.concurrent.Future
+import scala.concurrent.{ExecutionContext, Future}
 
 /**
   * This trait provides methods to query various types of blockchain data.
@@ -19,6 +19,14 @@ trait ChainQueryApi {
   /** Gets the hash of the block that is what we consider "best" */
   def getBestBlockHash(): Future[DoubleSha256DigestBE]
 
+  def getBestHashBlockHeight()(implicit ec: ExecutionContext): Future[Int] =
+    for {
+      hash <- getBestBlockHash()
+      heightOpt <- getBlockHeight(hash)
+      _ = require(heightOpt.isDefined,
+                  s"Best block hash must have a height! blockhash=$hash")
+    } yield heightOpt.get
+
   /** Gets number of confirmations for the given block hash*/
   def getNumberOfConfirmations(
       blockHashOpt: DoubleSha256DigestBE): Future[Option[Int]]

core/src/main/scala/org/bitcoins/core/api/NodeChainQueryApi.scala

@@ -0,0 +1,3 @@
+package org.bitcoins.core.api
+
+case class NodeChainQueryApi(nodeApi: NodeApi, chainQueryApi: ChainQueryApi)

core/src/main/scala/org/bitcoins/core/protocol/BlockStamp.scala

@@ -15,6 +15,9 @@ sealed trait BlockStamp {
 }
 
 object BlockStamp {
+  val height0 = BlockHeight(0)
+  val height0Opt = Some(height0)
+
   case class InvalidBlockStamp(blockStamp: String)
       extends RuntimeException(s"Invalid blockstamp: ${blockStamp}")
 

core/src/main/scala/org/bitcoins/core/util/FutureUtil.scala

@@ -41,4 +41,27 @@ object FutureUtil {
         }
     }
   }
+
+  /** Takes elements, groups them into batches of 'batchSize' and then calls f on them.
+    * The next batch does not start executing until the first batch is finished
+    * */
+  def batchExecute[T, U](
+      elements: Vector[T],
+      f: Vector[T] => Future[U],
+      init: U,
+      batchSize: Int)(implicit ec: ExecutionContext): Future[U] = {
+    val initF = Future.successful(init)
+    val batches = elements.grouped(batchSize)
+    for {
+      batchExecution <- {
+        batches.foldLeft(initF) {
+          case (uF, batch) =>
+            for {
+              _ <- uF
+              executed <- f(batch)
+            } yield executed
+        }
+      }
+    } yield batchExecution
+  }
 }

docs/wallet/wallet-rescan.md

@@ -0,0 +1,89 @@
+---
+title: Wallet Rescans
+id: wallet-rescan
+---
+
+With [BIP157](https://github.com/bitcoin/bips/blob/master/bip-0157.mediawiki) you can cache block filters locally to use
+later for rescans in the case you need to restore your wallets. Our [chain](../applications/chain.md) project gives us
+an API with the ability to query for filters.
+
+You can rescan your wallet with filters with [`WalletApi.rescanNeutrinoWallet()`](https://github.com/bitcoin-s/bitcoin-s/blob/1a3b6b5b1e4eb8442dfab8b1a9faeff74418bdb0/wallet/src/main/scala/org/bitcoins/wallet/api/WalletApi.scala#L399)
+
+
+### Example
+
+To run this example you need to make sure you have access to a bitcoind binary. You can download this with bitcoin-s by doing
+`sbt downloadBitcoind`
+
+```scala mdoc:invisible
+import org.bitcoins.testkit.BitcoinSTestAppConfig
+import org.bitcoins.testkit.wallet._
+import org.bitcoins.server.BitcoinSAppConfig
+import akka.actor.ActorSystem
+import scala.concurrent.{ExecutionContext, Future, Await}
+import scala.concurrent.duration.DurationInt
+```
+
+```scala mdoc:compile-only
+
+//we need an actor system and app config to power this
+implicit val system: ActorSystem = ActorSystem(s"wallet-rescan-example")
+implicit val ec: ExecutionContext = system.dispatcher
+implicit val appConfig: BitcoinSAppConfig = BitcoinSTestAppConfig.getNeutrinoTestConfig()
+
+//ok now let's spin up a bitcoind and a bitcoin-s wallet with funds in it
+val walletWithBitcoindF = for {
+  w <- BitcoinSWalletTest.createWalletBitcoindNodeChainQueryApi()
+} yield w
+
+val walletF = walletWithBitcoindF.map(_.wallet)
+
+val bitcoindF = walletWithBitcoindF.map(_.bitcoind)
+
+//let's see what our initial wallet balance is
+val initBalanceF = for {
+  w <- walletF
+  balance <- w.getBalance()
+} yield {
+  println(s"Initial wallet balance=${balance}")
+  balance
+}
+
+//ok great! We have money in the wallet to start,
+//now let's delete our internal tables that hold our utxos
+//and addresses so that we end up with a 0 balance
+val clearedWalletF = for {
+  w <- walletF
+  _ <- initBalanceF
+  clearedWallet <- w.clearUtxosAndAddresses()
+  zeroBalance <- clearedWallet.getBalance()
+} yield {
+  println(s"Balance after clearing utxos: ${zeroBalance}")
+  clearedWallet
+}
+
+//we need to pick how many addresses we want to generate off of our keychain
+//when doing a rescan, this means we are generating 100 addrsses
+//and then looking for matches. If we find a match, we generate _another_
+//100 fresh addresses and search those. We keep doing this until we find
+//100 addresses that do not contain a match.
+val addrBatchSize = 100
+//ok now that we have a cleared wallet, we need to rescan and find our fudns again!
+val rescannedBalanceF = for {
+  w <- clearedWalletF
+  _ <- w.fullRescanNeurinoWallet(addrBatchSize)
+  balanceAfterRescan <- w.getBalance()
+} yield {
+  println(s"Wallet balance after rescan: ${balanceAfterRescan}")
+  ()
+}
+
+//cleanup
+val cleanupF = for {
+  _ <- rescannedBalanceF
+  walletWithBitcoind <- walletWithBitcoindF
+  _ <- BitcoinSWalletTest.destroyWalletWithBitcoind(walletWithBitcoind)
+} yield ()
+
+Await.result(cleanupF, 60.seconds)
+```

testkit/src/main/scala/org/bitcoins/testkit/chain/SyncUtil.scala

@@ -1,17 +1,22 @@
 package org.bitcoins.testkit.chain
 
 import org.bitcoins.chain.blockchain.sync.FilterWithHeaderHash
-import org.bitcoins.core.crypto.DoubleSha256DigestBE
-import org.bitcoins.core.gcs.{FilterType, GolombFilter}
-import org.bitcoins.core.protocol.blockchain.BlockHeader
+import org.bitcoins.core.api.ChainQueryApi.FilterResponse
+import org.bitcoins.core.api.{ChainQueryApi, NodeApi, NodeChainQueryApi}
+import org.bitcoins.core.crypto.{DoubleSha256Digest, DoubleSha256DigestBE}
+import org.bitcoins.core.gcs.{FilterType}
+import org.bitcoins.core.protocol.BlockStamp
+import org.bitcoins.core.protocol.blockchain.{Block, BlockHeader}
+import org.bitcoins.core.util.{BitcoinSLogger, FutureUtil}
 import org.bitcoins.rpc.client.common.BitcoindRpcClient
 import org.bitcoins.rpc.client.v19.BitcoindV19RpcClient
 import org.bitcoins.rpc.jsonmodels.GetBlockFilterResult
+import org.bitcoins.wallet.Wallet
 
 import scala.concurrent.{ExecutionContext, Future}
 
 /** Useful utilities to use in the chain project for syncing things against bitcoind */
-abstract class SyncUtil {
+abstract class SyncUtil extends BitcoinSLogger {
 
   /** Creates a function that will retrun bitcoin's best block hash when called */
   def getBestBlockHashFunc(
@@ -38,6 +43,178 @@ abstract class SyncUtil {
           FilterWithHeaderHash(filter, header)
       }
   }
+
+  def getChainQueryApi(bitcoindV19RpcClient: BitcoindV19RpcClient)(
+      implicit ec: ExecutionContext): ChainQueryApi = {
+    new ChainQueryApi {
+
+      /** Gets the height of the given block */
+      override def getBlockHeight(
+          blockHash: DoubleSha256DigestBE): Future[Option[Int]] = {
+        bitcoindV19RpcClient
+          .getBlockHeader(blockHash)
+          .map(b => Some(b.height))
+      }
+
+      /** Gets the hash of the block that is what we consider "best" */
+      override def getBestBlockHash(): Future[DoubleSha256DigestBE] = {
+        bitcoindV19RpcClient.getBestBlockHash
+      }
+
+      /** Gets number of confirmations for the given block hash */
+      override def getNumberOfConfirmations(
+          blockHashOpt: DoubleSha256DigestBE): Future[Option[Int]] = {
+        bitcoindV19RpcClient.getBlock(blockHashOpt).map { b =>
+          Some(b.confirmations)
+        }
+      }
+
+      /** Gets the number of compact filters in the database */
+      override def getFilterCount: Future[Int] = {
+        //filter count should be same as block height?
+        bitcoindV19RpcClient.getBlockCount
+      }
+
+      /** Returns the block height of the given block stamp */
+      override def getHeightByBlockStamp(
+          blockStamp: BlockStamp): Future[Int] = {
+        blockStamp match {
+          case BlockStamp.BlockHash(hash) => getBlockHeight(hash).map(_.get)
+          case BlockStamp.BlockHeight(height) =>
+            Future.successful(height)
+          case BlockStamp.BlockTime(_) =>
+            Future.failed(new RuntimeException(s"Cannot query by block time"))
+        }
+      }
+
+      override def getFiltersBetweenHeights(
+          startHeight: Int,
+          endHeight: Int): Future[Vector[FilterResponse]] = {
+        val allHeights = startHeight.to(endHeight)
+
+        def f(range: Vector[Int]): Future[Vector[FilterResponse]] = {
+          val filterFs = range.map { height =>
+            for {
+              hash <- bitcoindV19RpcClient.getBlockHash(height)
+              filter <- bitcoindV19RpcClient.getBlockFilter(hash,
+                                                            FilterType.Basic)
+            } yield {
+              FilterResponse(filter.filter, hash, height)
+            }
+          }
+          Future.sequence(filterFs)
+        }
+
+        FutureUtil.batchExecute(elements = allHeights.toVector,
+                                f = f,
+                                init = Vector.empty,
+                                batchSize = 25)
+      }
+    }
+  }
+
+  def getNodeApi(bitcoindRpcClient: BitcoindRpcClient)(
+      implicit ec: ExecutionContext): NodeApi = {
+    new NodeApi {
+
+      /**
+        * Request the underlying node to download the given blocks from its peers and feed the blocks to [[org.bitcoins.node.NodeCallbacks]].
+        */
+      override def downloadBlocks(
+          blockHashes: Vector[DoubleSha256Digest]): Future[Unit] = {
+        logger.info(s"Fetching ${blockHashes.length} hashes from bitcoind")
+        val f: Vector[DoubleSha256Digest] => Future[Vector[Unit]] = {
+          case hashes =>
+            val blocks: Vector[Future[Unit]] = hashes.map {
+              bitcoindRpcClient
+                .getBlockRaw(_)
+                .map(_ => ())
+            }
+            Future.sequence(blocks)
+        }
+
+        val batchSize = 25
+        val batchedExecutedF = FutureUtil.batchExecute(elements = blockHashes,
+                                                       f = f,
+                                                       init = Vector.empty,
+                                                       batchSize = batchSize)
+
+        batchedExecutedF.map { _ =>
+          logger.info(
+            s"Done fetching ${blockHashes.length} hashes from bitcoind")
+          ()
+        }
+      }
+    }
+  }
+
+  def getNodeApiWalletCallback(
+      bitcoindRpcClient: BitcoindRpcClient,
+      walletF: Future[Wallet])(implicit ec: ExecutionContext): NodeApi = {
+    new NodeApi {
+
+      /**
+        * Request the underlying node to download the given blocks from its peers and feed the blocks to [[org.bitcoins.node.NodeCallbacks]].
+        */
+      /**
+        * Request the underlying node to download the given blocks from its peers and feed the blocks to [[org.bitcoins.node.NodeCallbacks]].
+        */
+      override def downloadBlocks(
+          blockHashes: Vector[DoubleSha256Digest]): Future[Unit] = {
+        logger.info(s"Fetching ${blockHashes.length} hashes from bitcoind")
+        val f: Vector[DoubleSha256Digest] => Future[Wallet] = {
+          case hashes =>
+            val fetchedBlocks: Vector[Future[Block]] = hashes.map {
+              bitcoindRpcClient
+                .getBlockRaw(_)
+            }
+            val blocksF = Future.sequence(fetchedBlocks)
+
+            val updatedWalletF = for {
+              blocks <- blocksF
+              wallet <- walletF
+              processedWallet <- {
+                FutureUtil.foldLeftAsync(wallet, blocks) {
+                  case (wallet, block) =>
+                    wallet.processBlock(block).map(_.asInstanceOf[Wallet])
+                }
+              }
+            } yield processedWallet
+
+            updatedWalletF
+        }
+
+        val batchSize = 25
+        val batchedExecutedF = FutureUtil.batchExecute(elements = blockHashes,
+                                                       f = f,
+                                                       init = Vector.empty,
+                                                       batchSize = batchSize)
+
+        batchedExecutedF.map { _ =>
+          logger.info(
+            s"Done fetching ${blockHashes.length} hashes from bitcoind")
+          ()
+        }
+      }
+    }
+  }
+
+  def getNodeChainQueryApi(bitcoindV19RpcClient: BitcoindV19RpcClient)(
+      implicit ec: ExecutionContext): NodeChainQueryApi = {
+    val chainQuery = SyncUtil.getChainQueryApi(bitcoindV19RpcClient)
+    val nodeApi = SyncUtil.getNodeApi(bitcoindV19RpcClient)
+    NodeChainQueryApi(nodeApi, chainQuery)
+  }
+
+  def getNodeChainQueryApiWalletCallback(
+      bitcoindV19RpcClient: BitcoindV19RpcClient,
+      walletF: Future[Wallet])(
+      implicit ec: ExecutionContext): NodeChainQueryApi = {
+    val chainQuery = SyncUtil.getChainQueryApi(bitcoindV19RpcClient)
+    val nodeApi =
+      SyncUtil.getNodeApiWalletCallback(bitcoindV19RpcClient, walletF)
+    NodeChainQueryApi(nodeApi, chainQuery)
+  }
 }
 
 object SyncUtil extends SyncUtil

testkit/src/main/scala/org/bitcoins/testkit/node/NodeUnitTest.scala

@@ -289,8 +289,8 @@ object NodeUnitTest extends P2PLogger {
       appConfig: BitcoinSAppConfig): Future[Unit] = {
     import system.dispatcher
     val walletWithBitcoind = {
-      BitcoinSWalletTest.WalletWithBitcoind(fundedWalletBitcoind.wallet,
-                                            fundedWalletBitcoind.bitcoindRpc)
+      BitcoinSWalletTest.WalletWithBitcoindRpc(fundedWalletBitcoind.wallet,
+                                               fundedWalletBitcoind.bitcoindRpc)
     }
 
     //these need to be done in order, as the spv node needs to be