Fail broadcasting transaction when disconnected (#2336)

* Fail broadcasting transaction when disconnected

* Wait until disconnected

* Move test

* Move to separate file

* Attempt fix

* Make test never have a peer

* Improve reliablity

node-test/src/test/scala/org/bitcoins/node/BroadcastTransactionTest.scala

@@ -3,13 +3,12 @@ package org.bitcoins.node
 import org.bitcoins.core.currency._
 import org.bitcoins.core.protocol.transaction.Transaction
 import org.bitcoins.rpc.BitcoindException
+import org.bitcoins.rpc.client.v19.BitcoindV19RpcClient
 import org.bitcoins.server.BitcoinSAppConfig
 import org.bitcoins.testkit.BitcoinSTestAppConfig
-import org.bitcoins.testkit.Implicits._
 import org.bitcoins.testkit.async.TestAsyncUtil
-import org.bitcoins.testkit.core.gen.TransactionGenerators
 import org.bitcoins.testkit.node.NodeUnitTest
-import org.bitcoins.testkit.node.fixture.SpvNodeConnectedWithBitcoind
+import org.bitcoins.testkit.node.fixture.SpvNodeConnectedWithBitcoindV19
 import org.scalatest.FutureOutcome
 
 import scala.concurrent.Future
@@ -22,19 +21,28 @@ class BroadcastTransactionTest extends NodeUnitTest {
   implicit override protected def config: BitcoinSAppConfig =
     BitcoinSTestAppConfig.getSpvWithEmbeddedDbTestConfig(pgUrl)
 
-  override type FixtureParam = SpvNodeConnectedWithBitcoind
+  override type FixtureParam = SpvNodeConnectedWithBitcoindV19
 
   def withFixture(test: OneArgAsyncTest): FutureOutcome =
-    withSpvNodeConnectedToBitcoind(test)
+    withSpvNodeConnectedToBitcoindV19(test)
 
   private val sendAmount = 1.bitcoin
 
-  it must "safely broadcast a transaction twice" in { param =>
-    val node = param.node
+  def createValidTx(rpc: BitcoindV19RpcClient): Future[Transaction] = {
+    for {
+      rawTx <-
+        rpc.createRawTransaction(Vector.empty, Map(junkAddress -> sendAmount))
+      fundedTx <- rpc.fundRawTransaction(rawTx)
+      tx <- rpc.signRawTransactionWithWallet(fundedTx.hex).map(_.hex)
+    } yield tx
+  }
 
-    val tx = TransactionGenerators.transaction.sampleSome
+  it must "safely broadcast a transaction twice" in { param =>
+    val SpvNodeConnectedWithBitcoindV19(node, rpc) = param
 
     for {
+      tx <- createValidTx(rpc)
+
       _ <- node.broadcastTransaction(tx)
       _ <- node.broadcastTransaction(tx)
 
@@ -46,7 +54,7 @@ class BroadcastTransactionTest extends NodeUnitTest {
   }
 
   it must "broadcast a transaction" in { param =>
-    val SpvNodeConnectedWithBitcoind(node, rpc) = param
+    val SpvNodeConnectedWithBitcoindV19(node, rpc) = param
 
     def hasSeenTx(transaction: Transaction): Future[Boolean] = {
       rpc
@@ -78,14 +86,9 @@ class BroadcastTransactionTest extends NodeUnitTest {
     }
 
     for {
-      // fund bitcoind
-      _ <- rpc.getNewAddress.flatMap(rpc.generateToAddress(101, _))
       bitcoindBalancePreBroadcast <- rpc.getBalance
 
-      rawTx <-
-        rpc.createRawTransaction(Vector.empty, Map(junkAddress -> sendAmount))
-      fundedTx <- rpc.fundRawTransaction(rawTx)
-      tx <- rpc.signRawTransactionWithWallet(fundedTx.hex).map(_.hex)
+      tx <- createValidTx(rpc)
 
       _ <- attemptBroadcast(tx)
         .recoverWith {

node-test/src/test/scala/org/bitcoins/node/DisconnectedPeerTest.scala

@@ -0,0 +1,27 @@
+package org.bitcoins.node
+
+import org.bitcoins.server.BitcoinSAppConfig
+import org.bitcoins.testkit.BitcoinSTestAppConfig
+import org.bitcoins.testkit.Implicits._
+import org.bitcoins.testkit.core.gen.TransactionGenerators
+import org.bitcoins.testkit.node.NodeUnitTest
+import org.scalatest.FutureOutcome
+
+class DisconnectedPeerTest extends NodeUnitTest {
+
+  /** Wallet config with data directory set to user temp directory */
+  implicit override protected val config: BitcoinSAppConfig =
+    BitcoinSTestAppConfig.getSpvWithEmbeddedDbTestConfig(pgUrl)
+
+  override type FixtureParam = SpvNode
+
+  def withFixture(test: OneArgAsyncTest): FutureOutcome =
+    withDisconnectedSpvNode(test)
+
+  it must "fail to broadcast a transaction when disconnected" in { node =>
+    val tx = TransactionGenerators.transaction.sampleSome
+    recoverToSucceededIf[RuntimeException] {
+      node.broadcastTransaction(tx)
+    }
+  }
+}

node/src/main/scala/org/bitcoins/node/Node.scala

@@ -229,10 +229,19 @@ trait Node extends NodeApi with ChainQueryApi with P2PLogger {
       // message because a Bitcoin-S node currently doesn't have a mempool and only
       // broadcasts/relays transactions from its own wallet.
       // See https://developer.bitcoin.org/reference/p2p_networking.html#tx
-      _ =
-        logger.info(s"Sending out tx message for tx=${transaction.txIdBE.hex}")
-      _ <- peerMsgSender.sendTransactionMessage(transaction)
-    } yield ()
+      connected <- isConnected
+
+      res <- {
+        if (connected) {
+          logger.info(
+            s"Sending out tx message for tx=${transaction.txIdBE.hex}")
+          peerMsgSender.sendTransactionMessage(transaction)
+        } else {
+          Future.failed(new RuntimeException(
+            s"Error broadcasting transaction ${transaction.txIdBE.hex}, peer is disconnected $peer"))
+        }
+      }
+    } yield res
   }
 
   /**

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

@@ -1,7 +1,6 @@
 package org.bitcoins.testkit.node
 
 import java.net.InetSocketAddress
-
 import akka.actor.{ActorSystem, Cancellable}
 import org.bitcoins.chain.config.ChainAppConfig
 import org.bitcoins.core.api.chain.{ChainApi, ChainQueryApi, FilterSyncMarker}
@@ -29,12 +28,14 @@ import org.bitcoins.node.networking.peer.{
 import org.bitcoins.rpc.client.common.BitcoindVersion.{V18, V19}
 import org.bitcoins.rpc.client.common.{BitcoindRpcClient, BitcoindVersion}
 import org.bitcoins.rpc.client.v19.BitcoindV19RpcClient
+import org.bitcoins.rpc.util.RpcUtil
 import org.bitcoins.server.BitcoinSAppConfig
 import org.bitcoins.server.BitcoinSAppConfig._
 import org.bitcoins.testkit.EmbeddedPg
 import org.bitcoins.testkit.chain.ChainUnitTest
 import org.bitcoins.testkit.fixtures.BitcoinSFixture
 import org.bitcoins.testkit.keymanager.KeyManagerTestUtil
+import org.bitcoins.testkit.node.NodeUnitTest.{createPeer, emptyPeer}
 import org.bitcoins.testkit.node.fixture.{
   NeutrinoNodeConnectedWithBitcoind,
   NodeConnectedWithBitcoind,
@@ -172,6 +173,32 @@ trait NodeUnitTest extends BitcoinSFixture with EmbeddedPg {
       Future.successful(0)
   }
 
+  def withDisconnectedSpvNode(test: OneArgAsyncTest)(implicit
+      system: ActorSystem,
+      appConfig: BitcoinSAppConfig): FutureOutcome = {
+
+    val nodeBuilder: () => Future[SpvNode] = { () =>
+      require(appConfig.nodeType == NodeType.SpvNode)
+      for {
+        node <- NodeUnitTest.createSpvNode(
+          emptyPeer,
+          NodeCallbacks.empty,
+          start = false)(system, appConfig.chainConf, appConfig.nodeConf)
+        _ <- appConfig.start()
+      } yield node
+    }
+
+    makeDependentFixture(
+      build = nodeBuilder,
+      destroy = (_: Node) => {
+        for {
+          _ <- ChainUnitTest.destroyAllTables()
+          _ <- appConfig.stop()
+        } yield ()
+      }
+    )(test)
+  }
+
   def withSpvNodeConnectedToBitcoind(
       test: OneArgAsyncTest,
       versionOpt: Option[BitcoindVersion] = None)(implicit
@@ -182,7 +209,8 @@ trait NodeUnitTest extends BitcoinSFixture with EmbeddedPg {
         require(appConfig.nodeType == NodeType.SpvNode)
         for {
           bitcoind <- BitcoinSFixture.createBitcoind(versionOpt)
-          node <- NodeUnitTest.createSpvNode(bitcoind, NodeCallbacks.empty)(
+          node <- NodeUnitTest.createSpvNode(createPeer(bitcoind),
+                                             NodeCallbacks.empty)(
             system,
             appConfig.chainConf,
             appConfig.nodeConf)
@@ -207,10 +235,9 @@ trait NodeUnitTest extends BitcoinSFixture with EmbeddedPg {
           BitcoinSFixture
             .createBitcoindWithFunds(Some(V19))
             .map(_.asInstanceOf[BitcoindV19RpcClient])
-        node <- NodeUnitTest.createSpvNode(bitcoind, NodeCallbacks.empty)(
-          system,
-          appConfig.chainConf,
-          appConfig.nodeConf)
+        node <- NodeUnitTest.createSpvNode(
+          createPeer(bitcoind),
+          NodeCallbacks.empty)(system, appConfig.chainConf, appConfig.nodeConf)
       } yield SpvNodeConnectedWithBitcoindV19(node, bitcoind)
     }
 
@@ -390,7 +417,7 @@ object NodeUnitTest extends P2PLogger {
     require(appConfig.nodeType == NodeType.SpvNode)
     for {
       bitcoind <- BitcoinSFixture.createBitcoindWithFunds(versionOpt)
-      node <- createSpvNode(bitcoind, nodeCallbacks)
+      node <- createSpvNode(createPeer(bitcoind), nodeCallbacks)
       fundedWallet <- BitcoinSWalletTest.fundedWalletAndBitcoind(
         bitcoind,
         node,
@@ -478,11 +505,18 @@ object NodeUnitTest extends P2PLogger {
     Peer(id = None, socket = socket)
   }
 
+  def emptyPeer: Peer = {
+    val socket = new InetSocketAddress(RpcUtil.randomPort)
+    Peer(id = None, socket = socket)
+  }
+
   /** Creates a spv node peered with the given bitcoind client, this method
     * also calls [[org.bitcoins.node.Node.start() start]] to start the node
     */
-  def createSpvNode(bitcoind: BitcoindRpcClient, callbacks: NodeCallbacks)(
-      implicit
+  def createSpvNode(
+      peer: Peer,
+      callbacks: NodeCallbacks,
+      start: Boolean = true)(implicit
       system: ActorSystem,
       chainAppConfig: ChainAppConfig,
       nodeAppConfig: NodeAppConfig): Future[SpvNode] = {
@@ -497,7 +531,6 @@ object NodeUnitTest extends P2PLogger {
       _ <- checkConfigF
       chainHandler <- ChainUnitTest.createChainHandler()
     } yield chainHandler
-    val peer = createPeer(bitcoind)
     val nodeF = for {
       _ <- chainApiF
     } yield {
@@ -510,7 +543,10 @@ object NodeUnitTest extends P2PLogger {
       ).setBloomFilter(NodeTestUtil.emptyBloomFilter)
     }
 
-    nodeF.flatMap(_.start()).flatMap(_ => nodeF)
+    if (start)
+      nodeF.flatMap(_.start()).flatMap(_ => nodeF)
+    else nodeF
+
   }
 
   /** Creates a Neutrino node peered with the given bitcoind client, this method