Fix tx_signatures retransmission (#2748)

* Always retransmit tx_signatures if requested

If our peer asks us to retransmit our `tx_signatures`, we should do it even
if the transaction is already confirmed. That means we need to store our
`tx_signatures` because they're annoying to recompute.

* Ignore previously received commit_sig

When expecting a retransmission of `tx_signatures`, we should ignore the
`commit_sig` they send just before if we've already received it. The right
way to check that we've already received it is to compare its signature
to our latest commitment transaction.

eclair-core/src/main/scala/fr/acinq/eclair/channel/ChannelData.scala

@@ -27,7 +27,7 @@ import fr.acinq.eclair.io.Peer
 import fr.acinq.eclair.payment.OutgoingPaymentPacket.Upstream
 import fr.acinq.eclair.transactions.CommitmentSpec
 import fr.acinq.eclair.transactions.Transactions._
-import fr.acinq.eclair.wire.protocol.{ChannelAnnouncement, ChannelReady, ChannelReestablish, ChannelUpdate, ClosingSigned, CommitSig, FailureMessage, FundingCreated, FundingSigned, Init, OnionRoutingPacket, OpenChannel, OpenDualFundedChannel, Shutdown, SpliceInit, Stfu, UpdateAddHtlc, UpdateFailHtlc, UpdateFailMalformedHtlc, UpdateFulfillHtlc}
+import fr.acinq.eclair.wire.protocol.{ChannelAnnouncement, ChannelReady, ChannelReestablish, ChannelUpdate, ClosingSigned, CommitSig, FailureMessage, FundingCreated, FundingSigned, Init, OnionRoutingPacket, OpenChannel, OpenDualFundedChannel, Shutdown, SpliceInit, Stfu, TxSignatures, UpdateAddHtlc, UpdateFailHtlc, UpdateFailMalformedHtlc, UpdateFulfillHtlc}
 import fr.acinq.eclair.{Alias, BlockHeight, CltvExpiry, CltvExpiryDelta, Features, InitFeature, MilliSatoshi, MilliSatoshiLong, RealShortChannelId, UInt64}
 import scodec.bits.ByteVector
 
@@ -412,7 +412,11 @@ object RealScidStatus {
  */
 case class ShortIds(real: RealScidStatus, localAlias: Alias, remoteAlias_opt: Option[Alias])
 
-sealed trait LocalFundingStatus { def signedTx_opt: Option[Transaction] }
+sealed trait LocalFundingStatus {
+  def signedTx_opt: Option[Transaction]
+  /** We store local signatures for the purpose of retransmitting if the funding/splicing flow is interrupted. */
+  def localSigs_opt: Option[TxSignatures]
+}
 object LocalFundingStatus {
   sealed trait NotLocked extends LocalFundingStatus
   sealed trait Locked extends LocalFundingStatus
@@ -424,14 +428,17 @@ object LocalFundingStatus {
    * didn't keep the funding tx at all, even as funder (e.g. NORMAL). However, right after restoring those channels we
    * retrieve the funding tx and update the funding status immediately.
    */
-  case class SingleFundedUnconfirmedFundingTx(signedTx_opt: Option[Transaction]) extends UnconfirmedFundingTx with NotLocked
+  case class SingleFundedUnconfirmedFundingTx(signedTx_opt: Option[Transaction]) extends UnconfirmedFundingTx with NotLocked {
+    override val localSigs_opt: Option[TxSignatures] = None
+  }
   case class DualFundedUnconfirmedFundingTx(sharedTx: SignedSharedTransaction, createdAt: BlockHeight, fundingParams: InteractiveTxParams) extends UnconfirmedFundingTx with NotLocked {
-    override def signedTx_opt: Option[Transaction] = sharedTx.signedTx_opt
+    override val signedTx_opt: Option[Transaction] = sharedTx.signedTx_opt
+    override val localSigs_opt: Option[TxSignatures] = Some(sharedTx.localSigs)
   }
-  case class ZeroconfPublishedFundingTx(tx: Transaction) extends UnconfirmedFundingTx with Locked {
+  case class ZeroconfPublishedFundingTx(tx: Transaction, localSigs_opt: Option[TxSignatures]) extends UnconfirmedFundingTx with Locked {
     override val signedTx_opt: Option[Transaction] = Some(tx)
   }
-  case class ConfirmedFundingTx(tx: Transaction) extends LocalFundingStatus with Locked {
+  case class ConfirmedFundingTx(tx: Transaction, localSigs_opt: Option[TxSignatures]) extends LocalFundingStatus with Locked {
     override val signedTx_opt: Option[Transaction] = Some(tx)
   }
 }

eclair-core/src/main/scala/fr/acinq/eclair/channel/Commitments.scala

@@ -1132,6 +1132,16 @@ case class Commitments(params: ChannelParams,
     }
   }
 
+  /** This function should be used to ignore a commit_sig that we've already received. */
+  def ignoreRetransmittedCommitSig(commitSig: CommitSig): Boolean = {
+    val latestRemoteSig = latest.localCommit.commitTxAndRemoteSig.remoteSig
+    params.channelFeatures.hasFeature(Features.DualFunding) && commitSig.batchSize == 1 && latestRemoteSig == commitSig.signature
+  }
+
+  def localFundingSigs(fundingTxId: ByteVector32): Option[TxSignatures] = {
+    all.find(_.fundingTxId == fundingTxId).flatMap(_.localFundingStatus.localSigs_opt)
+  }
+
   /**
    * Update the local/remote funding status
    *

eclair-core/src/main/scala/fr/acinq/eclair/channel/fsm/Channel.scala

@@ -573,10 +573,10 @@ class Channel(val nodeParams: NodeParams, val wallet: OnChainChannelFunder with
                     stay() using d1 storing() sending signingSession1.localSigs calling endQuiescence(d1)
                 }
               }
-            case _ if d.commitments.params.channelFeatures.hasFeature(Features.DualFunding) && d.commitments.latest.localFundingStatus.signedTx_opt.isEmpty && commit.batchSize == 1 =>
-              // The latest funding transaction is unconfirmed and we're missing our peer's tx_signatures: any commit_sig
-              // that we receive before that should be ignored, it's either a retransmission of a commit_sig we've already
-              // received or a bug that will eventually lead to a force-close anyway.
+            case _ if d.commitments.ignoreRetransmittedCommitSig(commit) =>
+              // We haven't received our peer's tx_signatures for the latest funding transaction and asked them to resend it on reconnection.
+              // They also resend their corresponding commit_sig, but we have already received it so we should ignore it.
+              // Note that the funding transaction may have confirmed while we were reconnecting.
               log.info("ignoring commit_sig, we're still waiting for tx_signatures")
               stay()
             case _ =>
@@ -1095,7 +1095,7 @@ class Channel(val nodeParams: NodeParams, val wallet: OnChainChannelFunder with
       }
 
     case Event(w: WatchPublishedTriggered, d: DATA_NORMAL) =>
-      val fundingStatus = LocalFundingStatus.ZeroconfPublishedFundingTx(w.tx)
+      val fundingStatus = LocalFundingStatus.ZeroconfPublishedFundingTx(w.tx, d.commitments.localFundingSigs(w.tx.txid))
       d.commitments.updateLocalFundingStatus(w.tx.txid, fundingStatus) match {
         case Right((commitments1, _)) =>
           watchFundingConfirmed(w.tx.txid, Some(nodeParams.channelConf.minDepthBlocks))
@@ -1909,7 +1909,7 @@ class Channel(val nodeParams: NodeParams, val wallet: OnChainChannelFunder with
               d.spliceStatus match {
                 case SpliceStatus.SpliceWaitingForSigs(signingSession) if signingSession.fundingTx.txId == fundingTxId =>
                   // We retransmit our commit_sig, and will send our tx_signatures once we've received their commit_sig.
-                  log.info(s"re-sending commit_sig for splice attempt with fundingTxIndex=${signingSession.fundingTxIndex} fundingTxId=${signingSession.fundingTx.txId}")
+                  log.info("re-sending commit_sig for splice attempt with fundingTxIndex={} fundingTxId={}", signingSession.fundingTxIndex, signingSession.fundingTx.txId)
                   val commitSig = signingSession.remoteCommit.sign(keyManager, d.commitments.params, signingSession.fundingTxIndex, signingSession.fundingParams.remoteFundingPubKey, signingSession.commitInput)
                   sendQueue = sendQueue :+ commitSig
                   d.spliceStatus
@@ -1919,18 +1919,17 @@ class Channel(val nodeParams: NodeParams, val wallet: OnChainChannelFunder with
                       dfu.sharedTx match {
                         case fundingTx: InteractiveTxBuilder.PartiallySignedSharedTransaction =>
                           // If we have not received their tx_signatures, we can't tell whether they had received our commit_sig, so we need to retransmit it
-                          log.info(s"re-sending commit_sig and tx_signatures for fundingTxIndex=${d.commitments.latest.fundingTxIndex} fundingTxId=${d.commitments.latest.fundingTxId}")
+                          log.info("re-sending commit_sig and tx_signatures for fundingTxIndex={} fundingTxId={}", d.commitments.latest.fundingTxIndex, d.commitments.latest.fundingTxId)
                           val commitSig = d.commitments.latest.remoteCommit.sign(keyManager, d.commitments.params, d.commitments.latest.fundingTxIndex, d.commitments.latest.remoteFundingPubKey, d.commitments.latest.commitInput)
                           sendQueue = sendQueue :+ commitSig :+ fundingTx.localSigs
                         case fundingTx: InteractiveTxBuilder.FullySignedSharedTransaction =>
-                          log.info(s"re-sending tx_signatures for fundingTxIndex=${d.commitments.latest.fundingTxIndex} fundingTxId=${d.commitments.latest.fundingTxId}")
+                          log.info("re-sending tx_signatures for fundingTxIndex={} fundingTxId={}", d.commitments.latest.fundingTxIndex, d.commitments.latest.fundingTxId)
                           sendQueue = sendQueue :+ fundingTx.localSigs
                       }
-                    case _ =>
-                      // The funding tx is published or confirmed, and they have not received our tx_signatures, but they must have received our commit_sig, otherwise they
-                      // would not have sent their tx_signatures and we would not have been able to publish the funding tx in the first place. We could in theory
-                      // recompute our tx_signatures, but instead we do nothing: they will be notified that the funding tx has confirmed.
-                      log.warning("cannot re-send tx_signatures for fundingTxId={}, transaction is already published or confirmed", fundingTxId)
+                    case fundingStatus =>
+                      // They have not received our tx_signatures, but they must have received our commit_sig, otherwise we would be in the case above.
+                      log.info("re-sending tx_signatures for fundingTxIndex={} fundingTxId={} (already published or confirmed)", d.commitments.latest.fundingTxIndex, d.commitments.latest.fundingTxId)
+                      sendQueue = sendQueue ++ fundingStatus.localSigs_opt.toSeq
                   }
                   d.spliceStatus
                 case _ =>
@@ -1951,7 +1950,7 @@ class Channel(val nodeParams: NodeParams, val wallet: OnChainChannelFunder with
           val spliceLocked = d.commitments.active
             .filter(c => c.fundingTxIndex > 0) // only consider splice txs
             .collectFirst { case c if c.localFundingStatus.isInstanceOf[LocalFundingStatus.Locked] =>
-              log.debug(s"re-sending splice_locked for fundingTxId=${c.fundingTxId}")
+              log.debug("re-sending splice_locked for fundingTxId={}", c.fundingTxId)
               SpliceLocked(d.channelId, c.fundingTxId.reverse)
             }
           sendQueue = sendQueue ++ spliceLocked
@@ -2182,11 +2181,11 @@ class Channel(val nodeParams: NodeParams, val wallet: OnChainChannelFunder with
       // slightly before us. In that case, the WatchConfirmed may trigger first, and it would be inefficient to let the
       // WatchPublished override our funding status: it will make us set a new WatchConfirmed that will instantly
       // trigger and rewrite the funding status again.
-      val alreadyConfirmed = d.commitments.active.map(_.localFundingStatus).collect { case LocalFundingStatus.ConfirmedFundingTx(tx) => tx }.exists(_.txid == w.tx.txid)
+      val alreadyConfirmed = d.commitments.active.map(_.localFundingStatus).collect { case LocalFundingStatus.ConfirmedFundingTx(tx, _) => tx }.exists(_.txid == w.tx.txid)
       if (alreadyConfirmed) {
         stay()
       } else {
-        val fundingStatus = LocalFundingStatus.ZeroconfPublishedFundingTx(w.tx)
+        val fundingStatus = LocalFundingStatus.ZeroconfPublishedFundingTx(w.tx, d.commitments.localFundingSigs(w.tx.txid))
         d.commitments.updateLocalFundingStatus(w.tx.txid, fundingStatus) match {
           case Right((commitments1, _)) =>
             log.info(s"zero-conf funding txid=${w.tx.txid} has been published")

eclair-core/src/main/scala/fr/acinq/eclair/channel/fsm/ChannelOpenDualFunded.scala

@@ -673,7 +673,7 @@ trait ChannelOpenDualFunded extends DualFundingHandlers with ErrorHandlers {
 
     case Event(w: WatchPublishedTriggered, d: DATA_WAIT_FOR_DUAL_FUNDING_CONFIRMED) =>
       log.info("funding txid={} was successfully published for zero-conf channelId={}", w.tx.txid, d.channelId)
-      val fundingStatus = LocalFundingStatus.ZeroconfPublishedFundingTx(w.tx)
+      val fundingStatus = LocalFundingStatus.ZeroconfPublishedFundingTx(w.tx, d.commitments.localFundingSigs(w.tx.txid))
       d.commitments.updateLocalFundingStatus(w.tx.txid, fundingStatus) match {
         case Right((commitments1, _)) =>
           // we still watch the funding tx for confirmation even if we can use the zero-conf channel right away

eclair-core/src/main/scala/fr/acinq/eclair/channel/fsm/ChannelOpenSingleFunded.scala

@@ -389,7 +389,7 @@ trait ChannelOpenSingleFunded extends SingleFundingHandlers with ErrorHandlers {
       stay() using d.copy(deferred = Some(remoteChannelReady)) // no need to store, they will re-send if we get disconnected
 
     case Event(w: WatchPublishedTriggered, d: DATA_WAIT_FOR_FUNDING_CONFIRMED) =>
-      val fundingStatus = LocalFundingStatus.ZeroconfPublishedFundingTx(w.tx)
+      val fundingStatus = LocalFundingStatus.ZeroconfPublishedFundingTx(w.tx, None)
       d.commitments.updateLocalFundingStatus(w.tx.txid, fundingStatus) match {
         case Right((commitments1, _)) =>
           log.info("funding txid={} was successfully published for zero-conf channelId={}", w.tx.txid, d.channelId)

eclair-core/src/main/scala/fr/acinq/eclair/channel/fsm/CommonFundingHandlers.scala

@@ -68,7 +68,7 @@ trait CommonFundingHandlers extends CommonHandlers {
         }
       case _ => () // in the dual-funding case, we have already verified the funding tx
     }
-    val fundingStatus = ConfirmedFundingTx(w.tx)
+    val fundingStatus = ConfirmedFundingTx(w.tx, d.commitments.localFundingSigs(w.tx.txid))
     context.system.eventStream.publish(TransactionConfirmed(d.channelId, remoteNodeId, w.tx))
     d.commitments.updateLocalFundingStatus(w.tx.txid, fundingStatus).map {
       case (commitments1, commitment) =>

eclair-core/src/main/scala/fr/acinq/eclair/wire/internal/channel/version4/ChannelCodecs4.scala

@@ -14,7 +14,7 @@ import fr.acinq.eclair.transactions.Transactions._
 import fr.acinq.eclair.transactions.{CommitmentSpec, DirectedHtlc, IncomingHtlc, OutgoingHtlc}
 import fr.acinq.eclair.wire.protocol.CommonCodecs._
 import fr.acinq.eclair.wire.protocol.LightningMessageCodecs._
-import fr.acinq.eclair.wire.protocol.{UpdateAddHtlc, UpdateMessage}
+import fr.acinq.eclair.wire.protocol.{TxSignatures, UpdateAddHtlc, UpdateMessage}
 import fr.acinq.eclair.{BlockHeight, FeatureSupport, Features, PermanentChannelFeature, channel}
 import scodec.bits.{BitVector, ByteVector}
 import scodec.codecs._
@@ -336,8 +336,10 @@ private[channel] object ChannelCodecs4 {
     val fundingTxStatusCodec: Codec[LocalFundingStatus] = discriminated[LocalFundingStatus].by(uint8)
       .typecase(0x01, optional(bool8, txCodec).as[SingleFundedUnconfirmedFundingTx])
       .typecase(0x02, dualFundedUnconfirmedFundingTxCodec)
-      .typecase(0x03, txCodec.as[ZeroconfPublishedFundingTx])
-      .typecase(0x04, txCodec.as[ConfirmedFundingTx])
+      .typecase(0x05, (txCodec :: optional(bool8, lengthDelimited(txSignaturesCodec))).as[ZeroconfPublishedFundingTx])
+      .typecase(0x06, (txCodec :: optional(bool8, lengthDelimited(txSignaturesCodec))).as[ConfirmedFundingTx])
+      .typecase(0x03, (txCodec :: provide(Option.empty[TxSignatures])).as[ZeroconfPublishedFundingTx])
+      .typecase(0x04, (txCodec :: provide(Option.empty[TxSignatures])).as[ConfirmedFundingTx])
 
     val remoteFundingStatusCodec: Codec[RemoteFundingStatus] = discriminated[RemoteFundingStatus].by(uint8)
       .typecase(0x01, provide(RemoteFundingStatus.NotLocked))

eclair-core/src/test/scala/fr/acinq/eclair/channel/states/e/NormalSplicesStateSpec.scala

@@ -974,11 +974,10 @@ class NormalSplicesStateSpec extends TestKitBaseClass with FixtureAnyFunSuiteLik
     bob2blockchain.expectWatchPublished(spliceTx.txid)
     bob2blockchain.expectMsgType[WatchFundingDeeplyBuried]
 
-    // Alice doesn't retransmit her tx_signatures because the funding transaction has already been published.
+    alice2bob.expectMsgType[TxSignatures]
+    alice2bob.forward(bob)
     assert(alice2bob.expectMsgType[SpliceLocked].fundingTxid == spliceTx.txid)
     alice2bob.forward(bob)
-    // Bob cannot publish the transaction, but it will eventually confirm because it was published by Alice.
-    bob2blockchain.expectNoMessage(100 millis)
     bob2alice.expectNoMessage(100 millis)
     bob ! WatchFundingConfirmedTriggered(BlockHeight(42), 0, spliceTx)
     bob2alice.expectMsgType[SpliceLocked]
@@ -987,6 +986,38 @@ class NormalSplicesStateSpec extends TestKitBaseClass with FixtureAnyFunSuiteLik
     awaitCond(bob.stateData.asInstanceOf[DATA_NORMAL].commitments.active.size == 1)
   }
 
+  test("disconnect (tx_signatures sent by alice, splice confirms while bob is offline)") { f =>
+    import f._
+
+    val sender = initiateSpliceWithoutSigs(f, spliceOut_opt = Some(SpliceOut(20_000 sat, defaultSpliceOutScriptPubKey)))
+    alice2bob.expectMsgType[CommitSig]
+    alice2bob.forward(bob)
+    bob2alice.expectMsgType[CommitSig]
+    bob2alice.forward(alice)
+    bob2alice.expectMsgType[TxSignatures]
+    bob2alice.forward(alice)
+    alice2bob.expectMsgType[TxSignatures] // Bob doesn't receive Alice's tx_signatures
+    sender.expectMsgType[RES_SPLICE]
+    awaitCond(alice.stateData.asInstanceOf[DATA_NORMAL].spliceStatus == SpliceStatus.NoSplice)
+
+    // The splice transaction confirms while Bob is offline.
+    val spliceTx = alice.stateData.asInstanceOf[DATA_NORMAL].commitments.latest.localFundingStatus.signedTx_opt.get
+    disconnect(f)
+    alice ! WatchFundingConfirmedTriggered(BlockHeight(42), 0, spliceTx)
+
+    val (channelReestablishAlice, channelReestablishBob) = reconnect(f, interceptFundingDeeplyBuried = false)
+    assert(channelReestablishAlice.nextFundingTxId_opt.isEmpty)
+    assert(channelReestablishBob.nextFundingTxId_opt.contains(spliceTx.txid))
+    bob2alice.expectNoMessage(100 millis)
+
+    // Bob receives Alice's tx_signatures, which completes the splice.
+    alice2bob.expectMsgType[TxSignatures]
+    alice2bob.forward(bob)
+    alice2bob.expectMsgType[SpliceLocked]
+    alice2bob.forward(bob)
+    awaitCond(bob.stateData.asInstanceOf[DATA_NORMAL].spliceStatus == SpliceStatus.NoSplice)
+  }
+
   test("don't resend splice_locked when zero-conf channel confirms", Tag(ChannelStateTestsTags.ZeroConf), Tag(ChannelStateTestsTags.AnchorOutputsZeroFeeHtlcTxs)) { f =>
     import f._