Rework closing channel balance computation (#3096)

We re-work the channel balance computation (`CheckBalance.scala`) used
to continuously monitor the overall balance of the node. We previously
assumed that all pending HTLCs would timeout on-chain, and monitored
our mempool to deduplicate unconfirmed transactions and RBF attempts.
But it is actually simpler to assume that all pending HTLCs will be
fulfilled, and keep counting them in our off-chain balance until the
spending transaction reaches `min-depth`.

We introduce a distinction between recently confirmed transactions and
deeply confirmed transactions in our on-chain balance: there will be
cases where we don't know yet whether we'll be able to gets funds back
and the information of pending on-chain funds lets us verify that our
overall balance is looking good. It always eventually converges to be
in our on-chain balance.

Fixes #3085

Author: t-bast

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

@@ -398,7 +398,7 @@ class Setup(val datadir: File,
 
       _ = for (i <- 0 until config.getInt("autoprobe-count")) yield system.actorOf(SimpleSupervisor.props(Autoprobe.props(nodeParams, router, paymentInitiator), s"payment-autoprobe-$i", SupervisorStrategy.Restart))
 
-      balanceActor = system.spawn(BalanceActor(nodeParams.db, bitcoinClient, channelsListener, nodeParams.balanceCheckInterval), name = "balance-actor")
+      balanceActor = system.spawn(BalanceActor(bitcoinClient, nodeParams.channelConf.minDepth, channelsListener, nodeParams.balanceCheckInterval), name = "balance-actor")
 
       postman = system.spawn(Behaviors.supervise(Postman(nodeParams, switchboard, router.toTyped, register, offerManager)).onFailure(typed.SupervisorStrategy.restart), name = "postman")
 

eclair-core/src/main/scala/fr/acinq/eclair/balance/BalanceActor.scala

@@ -1,3 +1,19 @@
+/*
+ * Copyright 2024 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.balance
 
 import akka.actor.typed.eventstream.EventStream
@@ -7,12 +23,11 @@ import fr.acinq.bitcoin.scalacompat.{Btc, ByteVector32, SatoshiLong}
 import fr.acinq.eclair.NotificationsLogger
 import fr.acinq.eclair.NotificationsLogger.NotifyNodeOperator
 import fr.acinq.eclair.balance.BalanceActor._
-import fr.acinq.eclair.balance.CheckBalance.{GlobalBalance, computeOffChainBalance}
+import fr.acinq.eclair.balance.CheckBalance.{GlobalBalance, OffChainBalance}
 import fr.acinq.eclair.balance.Monitoring.{Metrics, Tags}
 import fr.acinq.eclair.blockchain.bitcoind.rpc.BitcoinCoreClient
 import fr.acinq.eclair.blockchain.bitcoind.rpc.BitcoinCoreClient.Utxo
 import fr.acinq.eclair.channel.PersistentChannelData
-import fr.acinq.eclair.db.Databases
 
 import scala.concurrent.ExecutionContext
 import scala.concurrent.duration._
@@ -29,26 +44,26 @@ object BalanceActor {
   private final case class WrappedGlobalBalanceWithChannels(wrapped: Try[GlobalBalance], channelsCount: Int) extends Command
   // @formatter:on
 
-  def apply(db: Databases, bitcoinClient: BitcoinCoreClient, channelsListener: ActorRef[ChannelsListener.GetChannels], interval: FiniteDuration)(implicit ec: ExecutionContext): Behavior[Command] = {
+  def apply(bitcoinClient: BitcoinCoreClient, minDepth: Int, channelsListener: ActorRef[ChannelsListener.GetChannels], interval: FiniteDuration)(implicit ec: ExecutionContext): Behavior[Command] = {
     Behaviors.setup { context =>
       Behaviors.withTimers { timers =>
         timers.startTimerWithFixedDelay(TickBalance, interval)
-        new BalanceActor(context, db, bitcoinClient, channelsListener).apply(refBalance_opt = None, previousBalance_opt = None)
+        new BalanceActor(context, bitcoinClient, minDepth, channelsListener).apply(refBalance_opt = None, previousBalance_opt = None)
       }
     }
   }
 
 }
 
 private class BalanceActor(context: ActorContext[Command],
-                           db: Databases,
                            bitcoinClient: BitcoinCoreClient,
+                           minDepth: Int,
                            channelsListener: ActorRef[ChannelsListener.GetChannels])(implicit ec: ExecutionContext) {
 
   private val log = context.log
 
   /**
-   * @param refBalance_opt the reference balance computed once at startup, useful for telling if we are making or losing money overall
+   * @param refBalance_opt      the reference balance computed once at startup, useful for telling if we are making or losing money overall
    * @param previousBalance_opt the last computed balance, it is useful to make a detailed diff between two successive balance checks
    * @return
    */
@@ -65,7 +80,7 @@ private class BalanceActor(context: ActorContext[Command],
       Behaviors.same
     case WrappedChannels(res) =>
       val channelsCount = res.channels.size
-      context.pipeToSelf(CheckBalance.computeGlobalBalance(res.channels, db, bitcoinClient))(b => WrappedGlobalBalanceWithChannels(b, channelsCount))
+      context.pipeToSelf(CheckBalance.computeGlobalBalance(res.channels, bitcoinClient, minDepth))(b => WrappedGlobalBalanceWithChannels(b, channelsCount))
       Behaviors.same
     case WrappedGlobalBalanceWithChannels(res, channelsCount) =>
       res match {
@@ -89,61 +104,61 @@ private class BalanceActor(context: ActorContext[Command],
           }
           previousBalance_opt match {
             case Some(previousBalance) =>
+              // On-chain metrics:
               log.info("on-chain diff={}", balance.onChain.total - previousBalance.onChain.total)
-              val utxosBefore = previousBalance.onChain.confirmed ++ previousBalance.onChain.unconfirmed
-              val utxosAfter = balance.onChain.confirmed ++ balance.onChain.unconfirmed
-              val utxosAdded = utxosAfter -- utxosBefore.keys
-              val utxosRemoved = utxosBefore -- utxosAfter.keys
+              val utxosBefore = previousBalance.onChain.utxos.toSet
+              val utxosAfter = balance.onChain.utxos.toSet
+              val utxosAdded = utxosAfter -- utxosBefore
+              val utxosRemoved = utxosBefore -- utxosAfter
               utxosAdded
-                .toList.sortBy(-_._2)
-                .foreach { case (outPoint, amount) => log.info("+ utxo={} amount={}", outPoint, amount) }
+                .toList.sortBy(_.amount)
+                .foreach(utxo => log.info("+ utxo={} amount={}", utxo.outPoint, utxo.amount))
               utxosRemoved
-                .toList.sortBy(-_._2)
-                .foreach { case (outPoint, amount) => log.info("- utxo={} amount={}", outPoint, amount) }
-
+                .toList.sortBy(_.amount)
+                .foreach(utxo => log.info("- utxo={} amount={}", utxo.outPoint, utxo.amount))
+              // Off-chain metrics:
               log.info("off-chain diff={}", balance.offChain.total - previousBalance.offChain.total)
-              val offchainBalancesBefore = previousBalance.channels.view.mapValues(computeOffChainBalance(previousBalance.knownPreimages, _).total)
-              val offchainBalancesAfter = balance.channels.view.mapValues(computeOffChainBalance(balance.knownPreimages, _).total)
-              offchainBalancesAfter
-                .map { case (channelId, balanceAfter) => (channelId, balanceAfter - offchainBalancesBefore.getOrElse(channelId, Btc(0))) }
+              val offChainBalancesBefore = previousBalance.channels.view.mapValues(channel => OffChainBalance().addChannelBalance(channel).total)
+              val offChainBalancesAfter = balance.channels.view.mapValues(channel => OffChainBalance().addChannelBalance(channel).total)
+              offChainBalancesAfter
+                .map { case (channelId, balanceAfter) => (channelId, balanceAfter - offChainBalancesBefore.getOrElse(channelId, Btc(0))) }
                 .filter { case (_, balanceDiff) => balanceDiff > 0.sat }
                 .toList.sortBy(-_._2)
                 .foreach { case (channelId, balanceDiff) => log.info("+ channelId={} amount={}", channelId, balanceDiff) }
-              offchainBalancesBefore
-                .map { case (channelId, balanceBefore) => (channelId, balanceBefore - offchainBalancesAfter.getOrElse(channelId, Btc(0))) }
+              offChainBalancesBefore
+                .map { case (channelId, balanceBefore) => (channelId, balanceBefore - offChainBalancesAfter.getOrElse(channelId, Btc(0))) }
                 .filter { case (_, balanceDiff) => balanceDiff > 0.sat }
                 .toList.sortBy(-_._2)
                 .foreach { case (channelId, balanceDiff) => log.info("- channelId={} amount={}", channelId, balanceDiff) }
             case None => ()
           }
-
-          log.info("current balance: total={} onchain.confirmed={} onchain.unconfirmed={} offchain={}", balance.total.toDouble, balance.onChain.totalConfirmed.toDouble, balance.onChain.totalUnconfirmed.toDouble, balance.offChain.total.toDouble)
+          log.info("current balance: total={} onchain.deeply-confirmed={} onchain.recently-confirmed={} onchain.unconfirmed={} offchain={}", balance.total.toDouble, balance.onChain.totalDeeplyConfirmed.toDouble, balance.onChain.totalRecentlyConfirmed.toDouble, balance.onChain.totalUnconfirmed.toDouble, balance.offChain.total.toDouble)
           log.debug("current balance details: {}", balance)
           // This is a very rough estimation of the fee we would need to pay for a force-close with 5 pending HTLCs at 100 sat/byte.
           val perChannelFeeBumpingReserve = 50_000.sat
           // Instead of scaling this linearly with the number of channels we have, we use sqrt(channelsCount) to reflect
           // the fact that if you have channels with many peers, only a subset of these peers will likely be malicious.
           val estimatedFeeBumpingReserve = perChannelFeeBumpingReserve * Math.sqrt(channelsCount)
-          if (balance.onChain.totalConfirmed < estimatedFeeBumpingReserve) {
+          val totalConfirmedBalance = balance.onChain.totalDeeplyConfirmed + balance.onChain.totalRecentlyConfirmed
+          if (totalConfirmedBalance < estimatedFeeBumpingReserve) {
             val message =
-              s"""On-chain confirmed balance is low (${balance.onChain.totalConfirmed.toMilliBtc}): eclair may not be able to guarantee funds safety in case channels force-close.
+              s"""On-chain confirmed balance is low (${totalConfirmedBalance.toMilliBtc}): eclair may not be able to guarantee funds safety in case channels force-close.
                  |You have $channelsCount channels, which could cost $estimatedFeeBumpingReserve in fees if some of these channels are malicious.
                  |Please note that the value above is a very arbitrary estimation: the real cost depends on the feerate and the number of malicious channels.
                  |You should add more utxos to your bitcoin wallet to guarantee funds safety.
                  |""".stripMargin
             context.system.eventStream ! EventStream.Publish(NotifyNodeOperator(NotificationsLogger.Warning, message))
           }
           Metrics.GlobalBalance.withoutTags().update(balance.total.toMilliBtc.toDouble)
-          Metrics.GlobalBalanceDetailed.withTag(Tags.BalanceType, Tags.BalanceTypes.OnchainConfirmed).update(balance.onChain.totalConfirmed.toMilliBtc.toDouble)
-          Metrics.GlobalBalanceDetailed.withTag(Tags.BalanceType, Tags.BalanceTypes.OnchainUnconfirmed).update(balance.onChain.totalUnconfirmed.toMilliBtc.toDouble)
-          Metrics.GlobalBalanceDetailed.withTag(Tags.BalanceType, Tags.BalanceTypes.Offchain).withTag(Tags.OffchainState, Tags.OffchainStates.waitForFundingConfirmed).update(balance.offChain.waitForFundingConfirmed.toMilliBtc.toDouble)
-          Metrics.GlobalBalanceDetailed.withTag(Tags.BalanceType, Tags.BalanceTypes.Offchain).withTag(Tags.OffchainState, Tags.OffchainStates.waitForChannelReady).update(balance.offChain.waitForChannelReady.toMilliBtc.toDouble)
-          Metrics.GlobalBalanceDetailed.withTag(Tags.BalanceType, Tags.BalanceTypes.Offchain).withTag(Tags.OffchainState, Tags.OffchainStates.normal).update(balance.offChain.normal.total.toMilliBtc.toDouble)
-          Metrics.GlobalBalanceDetailed.withTag(Tags.BalanceType, Tags.BalanceTypes.Offchain).withTag(Tags.OffchainState, Tags.OffchainStates.shutdown).update(balance.offChain.shutdown.total.toMilliBtc.toDouble)
-          Metrics.GlobalBalanceDetailed.withTag(Tags.BalanceType, Tags.BalanceTypes.Offchain).withTag(Tags.OffchainState, Tags.OffchainStates.closingLocal).update(balance.offChain.closing.localCloseBalance.total.toMilliBtc.toDouble)
-          Metrics.GlobalBalanceDetailed.withTag(Tags.BalanceType, Tags.BalanceTypes.Offchain).withTag(Tags.OffchainState, Tags.OffchainStates.closingRemote).update(balance.offChain.closing.remoteCloseBalance.total.toMilliBtc.toDouble)
-          Metrics.GlobalBalanceDetailed.withTag(Tags.BalanceType, Tags.BalanceTypes.Offchain).withTag(Tags.OffchainState, Tags.OffchainStates.closingUnknown).update(balance.offChain.closing.unknownCloseBalance.total.toMilliBtc.toDouble)
-          Metrics.GlobalBalanceDetailed.withTag(Tags.BalanceType, Tags.BalanceTypes.Offchain).withTag(Tags.OffchainState, Tags.OffchainStates.waitForPublishFutureCommitment).update(balance.offChain.waitForPublishFutureCommitment.toMilliBtc.toDouble)
+          Metrics.GlobalBalanceDetailed.withTag(Tags.BalanceType, Tags.BalanceTypes.OnChainDeeplyConfirmed).update(balance.onChain.totalDeeplyConfirmed.toMilliBtc.toDouble)
+          Metrics.GlobalBalanceDetailed.withTag(Tags.BalanceType, Tags.BalanceTypes.OnChainRecentlyConfirmed).update(balance.onChain.totalRecentlyConfirmed.toMilliBtc.toDouble)
+          Metrics.GlobalBalanceDetailed.withTag(Tags.BalanceType, Tags.BalanceTypes.OnChainUnconfirmed).update(balance.onChain.totalUnconfirmed.toMilliBtc.toDouble)
+          Metrics.GlobalBalanceDetailed.withTag(Tags.BalanceType, Tags.BalanceTypes.OffChain).withTag(Tags.OffChainState, Tags.OffChainStates.waitForFundingConfirmed).update(balance.offChain.waitForFundingConfirmed.toMilliBtc.toDouble)
+          Metrics.GlobalBalanceDetailed.withTag(Tags.BalanceType, Tags.BalanceTypes.OffChain).withTag(Tags.OffChainState, Tags.OffChainStates.waitForChannelReady).update(balance.offChain.waitForChannelReady.toMilliBtc.toDouble)
+          Metrics.GlobalBalanceDetailed.withTag(Tags.BalanceType, Tags.BalanceTypes.OffChain).withTag(Tags.OffChainState, Tags.OffChainStates.normal).update(balance.offChain.normal.total.toMilliBtc.toDouble)
+          Metrics.GlobalBalanceDetailed.withTag(Tags.BalanceType, Tags.BalanceTypes.OffChain).withTag(Tags.OffChainState, Tags.OffChainStates.shutdown).update(balance.offChain.shutdown.total.toMilliBtc.toDouble)
+          Metrics.GlobalBalanceDetailed.withTag(Tags.BalanceType, Tags.BalanceTypes.OffChain).withTag(Tags.OffChainState, Tags.OffChainStates.closing).update(balance.offChain.closing.total.toMilliBtc.toDouble)
+          Metrics.GlobalBalanceDetailed.withTag(Tags.BalanceType, Tags.BalanceTypes.OffChain).withTag(Tags.OffChainState, Tags.OffChainStates.waitForPublishFutureCommitment).update(balance.offChain.waitForPublishFutureCommitment.toMilliBtc.toDouble)
           refBalance_opt match {
             case Some(refBalance) =>
               val normalizedValue = 100 + (if (refBalance.total.toSatoshi.toLong > 0) (balance.total.toSatoshi.toLong - refBalance.total.toSatoshi.toLong) * 1000D / refBalance.total.toSatoshi.toLong else 0)
@@ -162,7 +177,7 @@ private class BalanceActor(context: ActorContext[Command],
           Behaviors.same
       }
     case GetGlobalBalance(replyTo, channels) =>
-      CheckBalance.computeGlobalBalance(channels, db, bitcoinClient) onComplete (replyTo ! _)
+      CheckBalance.computeGlobalBalance(channels, bitcoinClient, minDepth) onComplete (replyTo ! _)
       Behaviors.same
   }
 }