Make publishTransaction idempotent (#711)

Bitcoin core returns an error `missing inputs (code: -25)` if the tx that we want to publish has already been published and its output have been spent. When we receive this error, we try to get the tx, in order to know if it is in the blockchain, or if its inputs were spent by another tx.

Note: If the outputs of the tx were still unspent, bitcoin core would return "transaction already in block chain (code: -27)" and this is already handled.

eclair-core/src/main/scala/fr/acinq/eclair/blockchain/bitcoind/ZmqWatcher.scala

@@ -190,7 +190,6 @@ class ZmqWatcher(client: ExtendedBitcoinClient)(implicit ec: ExecutionContext =
 
         import scala.concurrent.duration._
         after(3 seconds, context.system.scheduler)(Future.successful({})).map(x => publish(tx, isRetry = true))
-      case t: Throwable if t.getMessage.contains("(code: -27)") => () // "transaction already in block chain (code: -27)" ignore error
       case t: Throwable => log.error(s"cannot publish tx: reason=${t.getMessage} txid=${tx.txid} tx=$tx")
     }
   }

eclair-core/src/main/scala/fr/acinq/eclair/blockchain/bitcoind/rpc/ExtendedBitcoinClient.scala

@@ -106,13 +106,27 @@ class ExtendedBitcoinClient(val rpcClient: BitcoinJsonRPCClient) {
     future
   }
 
-  def publishTransaction(hex: String)(implicit ec: ExecutionContext): Future[String] =
-    rpcClient.invoke("sendrawtransaction", hex) collect {
-      case JString(txid) => txid
-    }
-
+  /**
+    * Publish a transaction on the bitcoin network.
+    *
+    * Note that this method is idempotent, meaning that if the tx was already published a long time ago, then this is
+    * considered a success even if bitcoin core rejects this new attempt.
+    *
+    * @param tx
+    * @param ec
+    * @return
+    */
   def publishTransaction(tx: Transaction)(implicit ec: ExecutionContext): Future[String] =
-    publishTransaction(tx.toString())
+    rpcClient.invoke("sendrawtransaction", tx.toString()) collect {
+      case JString(txid) => txid
+    } recoverWith {
+      case JsonRPCError(Error(-27, _)) =>
+        // "transaction already in block chain (code: -27)" ignore error
+        Future.successful(tx.txid.toString())
+      case e@JsonRPCError(Error(-25, _)) =>
+        // "missing inputs (code: -25)" it may be that the tx has already been published and its output spent
+        getRawTransaction(tx.txid.toString()).map { case _ => tx.txid.toString() }.recoverWith { case _ => Future.failed[String](e) }
+    }
 
   /**
     * We need this to compute absolute timeouts expressed in number of blocks (where getBlockCount would be equivalent

eclair-core/src/test/scala/fr/acinq/eclair/blockchain/bitcoind/ExtendedBitcoinClientSpec.scala

@@ -0,0 +1,118 @@
+/*
+ * Copyright 2018 ACINQ SAS
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package fr.acinq.eclair.blockchain.bitcoind
+
+import akka.actor.ActorSystem
+import akka.actor.Status.Failure
+import akka.pattern.pipe
+import akka.testkit.{TestKit, TestProbe}
+import com.typesafe.config.ConfigFactory
+import fr.acinq.bitcoin.Transaction
+import fr.acinq.eclair.blockchain.bitcoind.rpc.{BasicBitcoinJsonRPCClient, ExtendedBitcoinClient}
+import grizzled.slf4j.Logging
+import org.json4s.JsonAST._
+import org.json4s.{DefaultFormats, JString}
+import org.junit.runner.RunWith
+import org.scalatest.junit.JUnitRunner
+import org.scalatest.{BeforeAndAfterAll, FunSuiteLike}
+
+import scala.collection.JavaConversions._
+import scala.concurrent.ExecutionContext.Implicits.global
+
+@RunWith(classOf[JUnitRunner])
+class ExtendedBitcoinClientSpec extends TestKit(ActorSystem("test")) with BitcoindService with FunSuiteLike with BeforeAndAfterAll with Logging {
+
+  val commonConfig = ConfigFactory.parseMap(Map("eclair.chain" -> "regtest", "eclair.spv" -> false, "eclair.server.public-ips.1" -> "localhost", "eclair.bitcoind.port" -> 28333, "eclair.bitcoind.rpcport" -> 28332, "eclair.bitcoind.zmq" -> "tcp://127.0.0.1:28334", "eclair.router-broadcast-interval" -> "2 second", "eclair.auto-reconnect" -> false))
+  val config = ConfigFactory.load(commonConfig).getConfig("eclair")
+
+  implicit val formats = DefaultFormats
+
+  override def beforeAll(): Unit = {
+    startBitcoind()
+  }
+
+  override def afterAll(): Unit = {
+    stopBitcoind()
+  }
+
+  test("wait bitcoind ready") {
+    waitForBitcoindReady()
+  }
+
+  test("send transaction idempotent") {
+    val bitcoinClient = new BasicBitcoinJsonRPCClient(
+      user = config.getString("bitcoind.rpcuser"),
+      password = config.getString("bitcoind.rpcpassword"),
+      host = config.getString("bitcoind.host"),
+      port = config.getInt("bitcoind.rpcport"))
+
+    val sender = TestProbe()
+    bitcoinClient.invoke("getnewaddress").pipeTo(sender.ref)
+    val JString(address) = sender.expectMsgType[JString]
+    bitcoinClient.invoke("createrawtransaction", Array.empty, Map(address -> 6)).pipeTo(sender.ref)
+    val JString(noinputTx) = sender.expectMsgType[JString]
+    bitcoinClient.invoke("fundrawtransaction", noinputTx).pipeTo(sender.ref)
+    val json = sender.expectMsgType[JValue]
+    val JString(unsignedtx) = json \ "hex"
+    val JInt(changePos) = json \ "changepos"
+    bitcoinClient.invoke("signrawtransaction", unsignedtx).pipeTo(sender.ref)
+    val JString(signedTx) = sender.expectMsgType[JValue] \ "hex"
+    val tx = Transaction.read(signedTx)
+    val txid = tx.txid.toString()
+
+    // test starts here
+    val client = new ExtendedBitcoinClient(bitcoinClient)
+    // we publish it a first time
+    client.publishTransaction(tx).pipeTo(sender.ref)
+    sender.expectMsg(txid)
+    // we publish the tx a second time to test idempotence
+    client.publishTransaction(tx).pipeTo(sender.ref)
+    sender.expectMsg(txid)
+    // let's confirm the tx
+    bitcoinClient.invoke("generate", 1).pipeTo(sender.ref)
+    sender.expectMsgType[JValue]
+    // and publish the tx a third time to test idempotence
+    client.publishTransaction(tx).pipeTo(sender.ref)
+    sender.expectMsg(txid)
+
+    // now let's spent the output of the tx
+    val spendingTx = {
+      val pos = if (changePos == 0) 1 else 0
+      bitcoinClient.invoke("createrawtransaction", Array(Map("txid" -> txid, "vout" -> pos)), Map(address -> 5.99999)).pipeTo(sender.ref)
+      val JString(unsignedtx) = sender.expectMsgType[JValue]
+      bitcoinClient.invoke("signrawtransaction", unsignedtx).pipeTo(sender.ref)
+      val JString(signedTx) = sender.expectMsgType[JValue] \ "hex"
+      signedTx
+    }
+    bitcoinClient.invoke("sendrawtransaction", spendingTx).pipeTo(sender.ref)
+    val JString(spendingTxid) = sender.expectMsgType[JValue]
+
+    // and publish the tx a fourth time to test idempotence
+    client.publishTransaction(tx).pipeTo(sender.ref)
+    sender.expectMsg(txid)
+    // let's confirm the tx
+    bitcoinClient.invoke("generate", 1).pipeTo(sender.ref)
+    sender.expectMsgType[JValue]
+    // and publish the tx a fifth time to test idempotence
+    client.publishTransaction(tx).pipeTo(sender.ref)
+    sender.expectMsg(txid)
+
+    // this one should be rejected
+    client.publishTransaction(Transaction.read("02000000000101b9e2a3f518fd74e696d258fed3c78c43f84504e76c99212e01cf225083619acf00000000000d0199800136b34b00000000001600145464ce1e5967773922506e285780339d72423244040047304402206795df1fd93c285d9028c384aacf28b43679f1c3f40215fd7bd1abbfb816ee5a022047a25b8c128e692d4717b6dd7b805aa24ecbbd20cfd664ab37a5096577d4a15d014730440220770f44121ed0e71ec4b482dded976f2febd7500dfd084108e07f3ce1e85ec7f5022025b32dc0d551c47136ce41bfb80f5a10de95c0babb22a3ae2d38e6688b32fcb20147522102c2662ab3e4fa18a141d3be3317c6ee134aff10e6cd0a91282a25bf75c0481ebc2102e952dd98d79aa796289fa438e4fdeb06ed8589ff2a0f032b0cfcb4d7b564bc3252aea58d1120")).pipeTo(sender.ref)
+    sender.expectMsgType[Failure]
+  }
+}