Handle fulfill not acked upstream

eclair-core/src/main/resources/reference.conf

@@ -68,6 +68,11 @@ eclair {
   max-to-local-delay-blocks = 2016 // maximum number of blocks that we are ready to accept for our own delayed outputs (2016 ~ 2 weeks)
   mindepth-blocks = 3
   expiry-delta-blocks = 144
+  // When we receive the pre-image for an HTLC and want to fulfill it but the upstream peer stops responding, we want to
+  // avoid letting its HTLC-timeout transaction become enforceable on-chain (otherwise there is a race condition between
+  // our HTLC-success and their HTLC-timeout).
+  // We will close the channel when the HTLC-timeout will happen in less than this number.
+  fulfill-safety-before-timeout-blocks = 6
 
   fee-base-msat = 1000
   fee-proportional-millionths = 100 // fee charged per transferred satoshi in millionths of a satoshi (100 = 0.01%)

eclair-core/src/main/scala/fr/acinq/eclair/NodeParams.scala

@@ -50,6 +50,7 @@ case class NodeParams(keyManager: KeyManager,
                       maxHtlcValueInFlightMsat: UInt64,
                       maxAcceptedHtlcs: Int,
                       expiryDeltaBlocks: Int,
+                      fulfillSafetyBeforeTimeoutBlocks: Int,
                       htlcMinimumMsat: Int,
                       toRemoteDelayBlocks: Int,
                       maxToLocalDelayBlocks: Int,
@@ -149,14 +150,18 @@ object NodeParams {
     val offeredCLTV = config.getInt("to-remote-delay-blocks")
     require(maxToLocalCLTV <= Channel.MAX_TO_SELF_DELAY && offeredCLTV <= Channel.MAX_TO_SELF_DELAY, s"CLTV delay values too high, max is ${Channel.MAX_TO_SELF_DELAY}")
 
+    val expiryDeltaBlocks = config.getInt("expiry-delta-blocks")
+    val fulfillSafetyBeforeTimeoutBlocks = config.getInt("fulfill-safety-before-timeout-blocks")
+    require(fulfillSafetyBeforeTimeoutBlocks < expiryDeltaBlocks, "fulfill-safety-before-timeout-blocks must be smaller than expiry-delta-blocks")
+
     val nodeAlias = config.getString("node-alias")
     require(nodeAlias.getBytes("UTF-8").length <= 32, "invalid alias, too long (max allowed 32 bytes)")
 
     val overrideFeatures: Map[PublicKey, (ByteVector, ByteVector)] = config.getConfigList("override-features").map { e =>
       val p = PublicKey(ByteVector.fromValidHex(e.getString("nodeid")))
       val gf = ByteVector.fromValidHex(e.getString("global-features"))
       val lf = ByteVector.fromValidHex(e.getString("local-features"))
-      (p -> (gf, lf))
+      p -> (gf, lf)
     }.toMap
 
     val socksProxy_opt = if (config.getBoolean("socks5.enabled")) {
@@ -187,7 +192,8 @@ object NodeParams {
       dustLimitSatoshis = dustLimitSatoshis,
       maxHtlcValueInFlightMsat = UInt64(config.getLong("max-htlc-value-in-flight-msat")),
       maxAcceptedHtlcs = maxAcceptedHtlcs,
-      expiryDeltaBlocks = config.getInt("expiry-delta-blocks"),
+      expiryDeltaBlocks = expiryDeltaBlocks,
+      fulfillSafetyBeforeTimeoutBlocks = fulfillSafetyBeforeTimeoutBlocks,
       htlcMinimumMsat = config.getInt("htlc-minimum-msat"),
       toRemoteDelayBlocks = config.getInt("to-remote-delay-blocks"),
       maxToLocalDelayBlocks = config.getInt("max-to-local-delay-blocks"),

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

@@ -18,7 +18,7 @@ package fr.acinq.eclair.channel
 
 import fr.acinq.bitcoin.Crypto.PrivateKey
 import fr.acinq.bitcoin.{ByteVector32, Transaction}
-import fr.acinq.eclair.{ShortChannelId, UInt64}
+import fr.acinq.eclair.UInt64
 import fr.acinq.eclair.payment.Origin
 import fr.acinq.eclair.wire.{ChannelUpdate, UpdateAddHtlc}
 
@@ -27,6 +27,7 @@ import fr.acinq.eclair.wire.{ChannelUpdate, UpdateAddHtlc}
   */
 
 class ChannelException(val channelId: ByteVector32, message: String) extends RuntimeException(message)
+
 // @formatter:off
 case class DebugTriggeredException             (override val channelId: ByteVector32) extends ChannelException(channelId, "debug-mode triggered failure")
 case class InvalidChainHash                    (override val channelId: ByteVector32, local: ByteVector32, remote: ByteVector32) extends ChannelException(channelId, s"invalid chainHash (local=$local remote=$remote)")
@@ -49,6 +50,7 @@ case class InvalidFinalScript                  (override val channelId: ByteVect
 case class FundingTxTimedout                   (override val channelId: ByteVector32) extends ChannelException(channelId, "funding tx timed out")
 case class FundingTxSpent                      (override val channelId: ByteVector32, spendingTx: Transaction) extends ChannelException(channelId, s"funding tx has been spent by txid=${spendingTx.txid}")
 case class HtlcTimedout                        (override val channelId: ByteVector32, htlcs: Set[UpdateAddHtlc]) extends ChannelException(channelId, s"one or more htlcs timed out: ids=${htlcs.take(10).map(_.id).mkString}") // we only display the first 10 ids
+case class HtlcWillTimeoutUpstream             (override val channelId: ByteVector32, htlcs: Set[UpdateAddHtlc]) extends ChannelException(channelId, s"one or more htlcs that should be fulfilled are close to timing out upstream: ids=${htlcs.take(10).map(_.id).mkString}") // we only display the first 10 ids
 case class HtlcOverridenByLocalCommit          (override val channelId: ByteVector32) extends ChannelException(channelId, "htlc was overriden by local commit")
 case class FeerateTooSmall                     (override val channelId: ByteVector32, remoteFeeratePerKw: Long) extends ChannelException(channelId, s"remote fee rate is too small: remoteFeeratePerKw=$remoteFeeratePerKw")
 case class FeerateTooDifferent                 (override val channelId: ByteVector32, localFeeratePerKw: Long, remoteFeeratePerKw: Long) extends ChannelException(channelId, s"local/remote feerates are too different: remoteFeeratePerKw=$remoteFeeratePerKw localFeeratePerKw=$localFeeratePerKw")

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

@@ -62,11 +62,23 @@ case class Commitments(channelVersion: ChannelVersion,
 
   def hasNoPendingHtlcs: Boolean = localCommit.spec.htlcs.isEmpty && remoteCommit.spec.htlcs.isEmpty && remoteNextCommitInfo.isRight
 
-  def timedoutOutgoingHtlcs(blockheight: Long): Set[UpdateAddHtlc] =
+  def timedOutOutgoingHtlcs(blockheight: Long): Set[UpdateAddHtlc] =
     (localCommit.spec.htlcs.filter(htlc => htlc.direction == OUT && blockheight >= htlc.add.cltvExpiry) ++
       remoteCommit.spec.htlcs.filter(htlc => htlc.direction == IN && blockheight >= htlc.add.cltvExpiry) ++
       remoteNextCommitInfo.left.toOption.map(_.nextRemoteCommit.spec.htlcs.filter(htlc => htlc.direction == IN && blockheight >= htlc.add.cltvExpiry)).getOrElse(Set.empty[DirectedHtlc])).map(_.add)
 
+  /**
+    * HTLCs that are close to timing out upstream are potentially dangerous. If we received the pre-image for those
+    * HTLCs, we need to get a remote signed updated commitment that removes this HTLC.
+    * Otherwise when we get close to the upstream timeout, we risk an on-chain race condition between their HTLC timeout
+    * and our HTLC success in case of a force-close.
+    */
+  def almostTimedOutIncomingHtlcs(blockheight: Long, fulfillSafety: Int): Set[UpdateAddHtlc] = {
+    localCommit.spec.htlcs.collect {
+      case htlc if htlc.direction == IN && blockheight >= htlc.add.cltvExpiry - fulfillSafety => htlc.add
+    }
+  }
+
   def addLocalProposal(proposal: UpdateMessage): Commitments = Commitments.addLocalProposal(this, proposal)
 
   def addRemoteProposal(proposal: UpdateMessage): Commitments = Commitments.addRemoteProposal(this, proposal)
@@ -87,12 +99,13 @@ case class Commitments(channelVersion: ChannelVersion,
 }
 
 object Commitments {
+
   /**
-    * add a change to our proposed change list
+    * Add a change to our proposed change list.
     *
-    * @param commitments
-    * @param proposal
-    * @return an updated commitment instance
+    * @param commitments current commitments.
+    * @param proposal    proposed change to add.
+    * @return an updated commitment instance.
     */
   private def addLocalProposal(commitments: Commitments, proposal: UpdateMessage): Commitments =
     commitments.copy(localChanges = commitments.localChanges.copy(proposed = commitments.localChanges.proposed :+ proposal))
@@ -212,14 +225,14 @@ object Commitments {
         val fulfill = UpdateFulfillHtlc(commitments.channelId, cmd.id, cmd.r)
         val commitments1 = addLocalProposal(commitments, fulfill)
         (commitments1, fulfill)
-      case Some(htlc) => throw InvalidHtlcPreimage(commitments.channelId, cmd.id)
+      case Some(_) => throw InvalidHtlcPreimage(commitments.channelId, cmd.id)
       case None => throw UnknownHtlcId(commitments.channelId, cmd.id)
     }
 
   def receiveFulfill(commitments: Commitments, fulfill: UpdateFulfillHtlc): Either[Commitments, (Commitments, Origin, UpdateAddHtlc)] =
     getHtlcCrossSigned(commitments, OUT, fulfill.id) match {
       case Some(htlc) if htlc.paymentHash == sha256(fulfill.paymentPreimage) => Right((addRemoteProposal(commitments, fulfill), commitments.originChannels(fulfill.id), htlc))
-      case Some(htlc) => throw InvalidHtlcPreimage(commitments.channelId, fulfill.id)
+      case Some(_) => throw InvalidHtlcPreimage(commitments.channelId, fulfill.id)
       case None => throw UnknownHtlcId(commitments.channelId, fulfill.id)
     }
 
@@ -244,7 +257,7 @@ object Commitments {
             val fail = UpdateFailHtlc(commitments.channelId, cmd.id, reason)
             val commitments1 = addLocalProposal(commitments, fail)
             (commitments1, fail)
-          case Failure(_) => throw new CannotExtractSharedSecret(commitments.channelId, htlc)
+          case Failure(_) => throw CannotExtractSharedSecret(commitments.channelId, htlc)
         }
       case None => throw UnknownHtlcId(commitments.channelId, cmd.id)
     }
@@ -263,7 +276,7 @@ object Commitments {
       } =>
         // we have already sent a fail/fulfill for this htlc
         throw UnknownHtlcId(commitments.channelId, cmd.id)
-      case Some(htlc) =>
+      case Some(_) =>
         val fail = UpdateFailMalformedHtlc(commitments.channelId, cmd.id, cmd.onionHash, cmd.failureCode)
         val commitments1 = addLocalProposal(commitments, fail)
         (commitments1, fail)
@@ -348,9 +361,9 @@ object Commitments {
 
   def remoteHasUnsignedOutgoingHtlcs(commitments: Commitments): Boolean = commitments.remoteChanges.proposed.collectFirst { case u: UpdateAddHtlc => u }.isDefined
 
-  def localHasChanges(commitments: Commitments): Boolean = commitments.remoteChanges.acked.size > 0 || commitments.localChanges.proposed.size > 0
+  def localHasChanges(commitments: Commitments): Boolean = commitments.remoteChanges.acked.nonEmpty || commitments.localChanges.proposed.nonEmpty
 
-  def remoteHasChanges(commitments: Commitments): Boolean = commitments.localChanges.acked.size > 0 || commitments.remoteChanges.proposed.size > 0
+  def remoteHasChanges(commitments: Commitments): Boolean = commitments.localChanges.acked.nonEmpty || commitments.remoteChanges.proposed.nonEmpty
 
   def revocationPreimage(seed: ByteVector32, index: Long): ByteVector32 = ShaChain.shaChainFromSeed(seed, 0xFFFFFFFFFFFFFFFFL - index)
 
@@ -428,21 +441,21 @@ object Commitments {
 
     val sortedHtlcTxs: Seq[TransactionWithInputInfo] = (htlcTimeoutTxs ++ htlcSuccessTxs).sortBy(_.input.outPoint.index)
     if (commit.htlcSignatures.size != sortedHtlcTxs.size) {
-      throw new HtlcSigCountMismatch(commitments.channelId, sortedHtlcTxs.size, commit.htlcSignatures.size)
+      throw HtlcSigCountMismatch(commitments.channelId, sortedHtlcTxs.size, commit.htlcSignatures.size)
     }
     val htlcSigs = sortedHtlcTxs.map(keyManager.sign(_, keyManager.htlcPoint(localParams.channelKeyPath), localPerCommitmentPoint))
     val remoteHtlcPubkey = Generators.derivePubKey(remoteParams.htlcBasepoint, localPerCommitmentPoint)
     // combine the sigs to make signed txes
     val htlcTxsAndSigs = (sortedHtlcTxs, htlcSigs, commit.htlcSignatures).zipped.toList.collect {
       case (htlcTx: HtlcTimeoutTx, localSig, remoteSig) =>
         if (Transactions.checkSpendable(Transactions.addSigs(htlcTx, localSig, remoteSig)).isFailure) {
-          throw new InvalidHtlcSignature(commitments.channelId, htlcTx.tx)
+          throw InvalidHtlcSignature(commitments.channelId, htlcTx.tx)
         }
         HtlcTxAndSigs(htlcTx, localSig, remoteSig)
       case (htlcTx: HtlcSuccessTx, localSig, remoteSig) =>
         // we can't check that htlc-success tx are spendable because we need the payment preimage; thus we only check the remote sig
-        if (Transactions.checkSig(htlcTx, remoteSig, remoteHtlcPubkey) == false) {
-          throw new InvalidHtlcSignature(commitments.channelId, htlcTx.tx)
+        if (!Transactions.checkSig(htlcTx, remoteSig, remoteHtlcPubkey)) {
+          throw InvalidHtlcSignature(commitments.channelId, htlcTx.tx)
         }
         HtlcTxAndSigs(htlcTx, localSig, remoteSig)
     }

eclair-core/src/test/scala/fr/acinq/eclair/TestConstants.scala

@@ -23,11 +23,11 @@ import fr.acinq.bitcoin.{Block, ByteVector32, Script}
 import fr.acinq.eclair.NodeParams.BITCOIND
 import fr.acinq.eclair.crypto.LocalKeyManager
 import fr.acinq.eclair.db._
-import fr.acinq.eclair.db.sqlite._
 import fr.acinq.eclair.io.Peer
 import fr.acinq.eclair.router.RouterConf
 import fr.acinq.eclair.wire.{Color, NodeAddress}
 import scodec.bits.ByteVector
+
 import scala.concurrent.duration._
 
 /**
@@ -42,7 +42,6 @@ object TestConstants {
 
   def inMemoryDb(connection: Connection = sqliteInMemory()): Databases = Databases.databaseByConnections(connection, connection, connection)
 
-
   object Alice {
     val seed = ByteVector32(ByteVector.fill(32)(1))
     val keyManager = new LocalKeyManager(seed, Block.RegtestGenesisBlock.hash)
@@ -60,6 +59,7 @@ object TestConstants {
       maxHtlcValueInFlightMsat = UInt64(150000000),
       maxAcceptedHtlcs = 100,
       expiryDeltaBlocks = 144,
+      fulfillSafetyBeforeTimeoutBlocks = 6,
       htlcMinimumMsat = 0,
       minDepthBlocks = 3,
       toRemoteDelayBlocks = 144,
@@ -69,7 +69,7 @@ object TestConstants {
       feeProportionalMillionth = 10,
       reserveToFundingRatio = 0.01, // note: not used (overridden below)
       maxReserveToFundingRatio = 0.05,
-      db = inMemoryDb(sqliteInMemory),
+      db = inMemoryDb(sqliteInMemory()),
       revocationTimeout = 20 seconds,
       pingInterval = 30 seconds,
       pingTimeout = 10 seconds,
@@ -126,6 +126,7 @@ object TestConstants {
       maxHtlcValueInFlightMsat = UInt64.MaxValue, // Bob has no limit on the combined max value of in-flight htlcs
       maxAcceptedHtlcs = 30,
       expiryDeltaBlocks = 144,
+      fulfillSafetyBeforeTimeoutBlocks = 6,
       htlcMinimumMsat = 1000,
       minDepthBlocks = 3,
       toRemoteDelayBlocks = 144,
@@ -135,7 +136,7 @@ object TestConstants {
       feeProportionalMillionth = 10,
       reserveToFundingRatio = 0.01, // note: not used (overridden below)
       maxReserveToFundingRatio = 0.05,
-      db = inMemoryDb(sqliteInMemory),
+      db = inMemoryDb(sqliteInMemory()),
       revocationTimeout = 20 seconds,
       pingInterval = 30 seconds,
       pingTimeout = 10 seconds,