2024 07 29 rm tx bitcoind callbacks (#5632)

* fix docs

* Fix docs

* WIP

* Integrate BitcoindCallbacks

* Revert loback-test.xml

* Remove logs

app/server/src/main/scala/org/bitcoins/server/BitcoinSServerMain.scala

@@ -28,18 +28,19 @@ import org.bitcoins.core.api.wallet.{NeutrinoHDWalletApi, WalletApi}
 import org.bitcoins.core.util.TimeUtil
 import org.bitcoins.dlc.node.DLCNode
 import org.bitcoins.dlc.node.config.DLCNodeAppConfig
-import org.bitcoins.dlc.wallet._
+import org.bitcoins.dlc.wallet.*
 import org.bitcoins.feeprovider.MempoolSpaceTarget.HourFeeTarget
-import org.bitcoins.feeprovider._
+import org.bitcoins.feeprovider.*
 import org.bitcoins.node.Node
 import org.bitcoins.node.config.NodeAppConfig
 import org.bitcoins.node.models.NodeStateDescriptorDAO
+import org.bitcoins.rpc.BitcoindCallbacks
 import org.bitcoins.rpc.BitcoindException.InWarmUp
 import org.bitcoins.rpc.client.common.BitcoindRpcClient
 import org.bitcoins.rpc.config.{BitcoindRpcAppConfig, ZmqConfig}
 import org.bitcoins.server.bitcoind.BitcoindSyncState
 import org.bitcoins.server.routes.{BitcoinSServerRunner, CommonRoutes, Server}
-import org.bitcoins.server.util._
+import org.bitcoins.server.util.*
 import org.bitcoins.tor.config.TorAppConfig
 import org.bitcoins.wallet.WalletHolder
 import org.bitcoins.wallet.config.WalletAppConfig
@@ -401,14 +402,6 @@ class BitcoinSServerMain(override val serverArgParser: ServerArgParser)(implicit
     } yield {
       WebsocketUtil.buildChainCallbacks(wsQueue, bitcoind)
     }
-    val nodeApiF = for {
-      bitcoind <- bitcoindF
-      chainCallbacks <- chainCallbacksF
-    } yield BitcoindRpcBackendUtil.buildBitcoindNodeApi(
-      bitcoind,
-      Future.successful(walletHolder),
-      Some(chainCallbacks)
-    )
 
     val feeProviderF = bitcoindF.map { bitcoind =>
       FeeProviderFactory.getFeeProviderOrElse(
@@ -423,13 +416,12 @@ class BitcoinSServerMain(override val serverArgParser: ServerArgParser)(implicit
     val loadWalletApiF = {
       for {
         bitcoind <- bitcoindF
-        nodeApi <- nodeApiF
         feeProvider <- feeProviderF
       } yield {
         val l = DLCWalletBitcoindBackendLoader(
           walletHolder = walletHolder,
           bitcoind = bitcoind,
-          nodeApi = nodeApi,
+          nodeApi = bitcoind,
           feeProvider = feeProvider
         )
 
@@ -447,6 +439,11 @@ class BitcoinSServerMain(override val serverArgParser: ServerArgParser)(implicit
           Some(walletName),
           conf.walletConf.aesPasswordOpt
         )
+        bitcoind <- bitcoindF
+        walletHolder = result._1
+        callback = BitcoindCallbacks.onBlockReceived(
+          walletHolder.processBlock(_).map(_ => ()))
+        _ = bitcoind.bitcoindRpcAppConfig.addCallbacks(callback)
       } yield result
     }
 
@@ -629,7 +626,8 @@ class BitcoinSServerMain(override val serverArgParser: ServerArgParser)(implicit
       pollingCancellable <- syncF.flatMap { _ =>
         if (bitcoindRpcConf.zmqConfig == ZmqConfig.empty) {
           val blockingPollingCancellable = BitcoindRpcBackendUtil
-            .startBitcoindBlockPolling(wallet, bitcoind, chainCallbacksOpt)
+            .startBitcoindBlockPolling(wallet, bitcoind, chainCallbacksOpt)(
+              nodeConf.callBacks.executeOnBlockReceivedCallbacks(_))
           val mempoolCancellable = BitcoindRpcBackendUtil
             .startBitcoindMempoolPolling(wallet, bitcoind) { tx =>
               nodeConf.callBacks

app/server/src/main/scala/org/bitcoins/server/BitcoindRpcBackendUtil.scala

@@ -443,7 +443,8 @@ object BitcoindRpcBackendUtil extends BitcoinSLogger {
       bitcoind: BitcoindRpcClient,
       chainCallbacksOpt: Option[ChainCallbacks],
       interval: FiniteDuration = 10.seconds
-  )(implicit system: ActorSystem): Cancellable = {
+  )(processBlock: Block => Future[Unit])(implicit
+      system: ActorSystem): Cancellable = {
     import system.dispatcher
 
     val processingBitcoindBlocks = new AtomicBoolean(false)
@@ -451,7 +452,6 @@ object BitcoindRpcBackendUtil extends BitcoinSLogger {
     system.scheduler.scheduleWithFixedDelay(0.seconds, interval) { () =>
       {
         val isBitcoindSyncedF = isBitcoindInSync(bitcoind)
-
         isBitcoindSyncedF.map { isBitcoindSynced =>
           if (!isBitcoindSynced) {
             logger.info(s"Bitcoind is not synced, waiting for IBD to complete.")
@@ -463,10 +463,10 @@ object BitcoindRpcBackendUtil extends BitcoinSLogger {
                 if (!rescanning) {
                   val pollFOptF =
                     pollBitcoind(
-                      wallet = wallet,
                       bitcoind = bitcoind,
                       chainCallbacksOpt = chainCallbacksOpt,
-                      prevCount = walletSyncState.height
+                      prevCount = walletSyncState.height,
+                      processBlock = processBlock
                     )
 
                   pollFOptF.flatMap {
@@ -484,9 +484,9 @@ object BitcoindRpcBackendUtil extends BitcoinSLogger {
             f.onComplete { _ =>
               processingBitcoindBlocks.set(false)
               BitcoindRpcBackendUtil.setSyncingFlag(
-                false,
-                bitcoind,
-                chainCallbacksOpt
+                syncing = false,
+                bitcoind = bitcoind,
+                chainCallbacksOpt = chainCallbacksOpt
               )
             } // reset polling variable
             f.failed.foreach(err =>
@@ -506,10 +506,10 @@ object BitcoindRpcBackendUtil extends BitcoinSLogger {
     *   completed when the sync is finished.
     */
   private def pollBitcoind(
-      wallet: WalletApi,
       bitcoind: BitcoindRpcClient,
       chainCallbacksOpt: Option[ChainCallbacks],
-      prevCount: Int
+      prevCount: Int,
+      processBlock: Block => Future[Unit]
   )(implicit system: ActorSystem): Future[Option[Future[Done]]] = {
     import system.dispatcher
     val atomicPrevCount = new AtomicInteger(prevCount)
@@ -523,7 +523,7 @@ object BitcoindRpcBackendUtil extends BitcoinSLogger {
     val processBlockSink: Sink[(Block, GetBlockHeaderResult), Future[Done]] = {
       Sink.foreachAsync[(Block, GetBlockHeaderResult)](1) {
         case (block, blockHeaderResult) =>
-          val processBlocksF = wallet.processBlock(block)
+          val processBlocksF = processBlock(block)
 
           processBlocksF.failed.foreach { case err =>
             val failedCount = atomicPrevCount.get

app/server/src/main/scala/org/bitcoins/server/util/CallbackUtil.scala

@@ -116,7 +116,20 @@ object CallbackUtil extends BitcoinSLogger {
         .runWith(txSink)
         .map(_ => ())
     }
-    val callbacks = NodeCallbacks(onTxReceived = Vector(onTx))
+
+    val blockSink = Sink.foreachAsync[Block](1) { block =>
+      wallet
+        .processBlock(block)
+        .map(_ => ())
+    }
+    val onBlock: OnBlockReceived = { block =>
+      Source
+        .single(block)
+        .runWith(blockSink)
+        .map(_ => ())
+    }
+    val callbacks = NodeCallbacks(onTxReceived = Vector(onTx),
+                                  onBlockReceived = Vector(onBlock))
     val streamManager = NodeCallbackStreamManager(callbacks)
     Future.successful(streamManager)
   }

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

@@ -38,7 +38,7 @@ import scala.concurrent.Future
   */
 class BitcoindRpcClient(override val instance: BitcoindInstance)(implicit
     override val system: ActorSystem,
-    bitcoindRpcAppConfig: BitcoindRpcAppConfig
+    val bitcoindRpcAppConfig: BitcoindRpcAppConfig
 ) extends Client
     with FeeRateApi
     with NodeApi

testkit-core/.jvm/src/main/resources/logback-test.xml

@@ -31,6 +31,10 @@
 
     <logger name="org.bitcoins.wallet" level="WARN"/>
 
+    <logger name="org.bitcoins.dlc.wallet" level="WARN"/>
+
+    <logger name="org.bitcoins.server" level="WARN"/>
+
     <!-- see what slick is compiling to in sql -->
     <logger name="slick" level="OFF"/>
 

wallet-test/src/test/scala/org/bitcoins/wallet/BitcoindBlockPollingTest.scala

@@ -52,7 +52,7 @@ class BitcoindBlockPollingTest
           bitcoind,
           None,
           1.second
-        )
+        )(wallet.processBlock(_).map(_ => ()))
         _ <- bitcoind.generateToAddress(6, bech32Address)
 
         // Wait for it to process