Reduce the number of RPC calls to bitcoind during force-close (#2902)

* Don't spawn anchor tx publisher if commit is confirmed

It is inefficient to spawn a tx publisher for anchor txs if we already
know that the commit tx is confirmed: we will make calls to our bitcoin
node that can easily be avoided. This can matter when force-closing a
large number of channels with frequent disconnections (e.g. wallets).

* Improve `TxTimeLocksMonitor` performance

When publishing a transaction that has CSV delays, we previously used
the watcher and set a `minDepth` on the parent transaction matching
the CSV delay of the child transaction. While this was very simple,
it was unnecessarily expensive for large CSV delays: the watcher would
check for tx confirmations at every block, even when the CSV delay is
very large. When we force-close a large number of channels, it results
in a very large number of RPC calls to our `bitcoind` node.

We don't use the watcher in the `TxTimeLocksMonitor` anymore: instead
we check the parent confirmations once, and then we check again after
the CSV delay.

* Add relative delay hints to `ZmqWatcher`

When we tell the `ZmqWatcher` to watch for confirmations on transactions
that have a relative delay, it is highly inefficient to call our bitcoin
node at every new block to check for confirmations (especially when the
parent transaction isn't even confirmed). We now tell the watcher about
the relative delay, which lets it check for confirmations only at block
heights where we expect the transaction to reach its minimum depth. This
is especially useful to improve performance for delayed transactions
that usually use a CSV of at least 720 blocks.

Author: t-bast

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

@@ -367,7 +367,7 @@ class Setup(val datadir: File,
       // we want to make sure the handler for post-restart broken HTLCs has finished initializing.
       _ <- postRestartCleanUpInitialized.future
 
-      txPublisherFactory = Channel.SimpleTxPublisherFactory(nodeParams, watcher, bitcoinClient)
+      txPublisherFactory = Channel.SimpleTxPublisherFactory(nodeParams, bitcoinClient)
       channelFactory = Peer.SimpleChannelFactory(nodeParams, watcher, relayer, bitcoinClient, txPublisherFactory)
       pendingChannelsRateLimiter = system.spawn(Behaviors.supervise(PendingChannelsRateLimiter(nodeParams, router.toTyped, channels)).onFailure(typed.SupervisorStrategy.resume), name = "pending-channels-rate-limiter")
       peerFactory = Switchboard.SimplePeerFactory(nodeParams, bitcoinClient, channelFactory, pendingChannelsRateLimiter, register, router.toTyped)

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

@@ -63,6 +63,7 @@ object ZmqWatcher {
   private case class PublishBlockHeight(current: BlockHeight) extends Command
   private case class ProcessNewBlock(blockId: BlockId) extends Command
   private case class ProcessNewTransaction(tx: Transaction) extends Command
+  private case class SetWatchHint(w: GenericWatch, hint: WatchHint) extends Command
 
   final case class ValidateRequest(replyTo: ActorRef[ValidateResult], ann: ChannelAnnouncement) extends Command
   final case class ValidateResult(c: ChannelAnnouncement, fundingTx: Either[Throwable, (Transaction, UtxoStatus)])
@@ -155,7 +156,8 @@ object ZmqWatcher {
   case class WatchFundingDeeplyBuried(replyTo: ActorRef[WatchFundingDeeplyBuriedTriggered], txId: TxId, minDepth: Long) extends WatchConfirmed[WatchFundingDeeplyBuriedTriggered]
   case class WatchFundingDeeplyBuriedTriggered(blockHeight: BlockHeight, txIndex: Int, tx: Transaction) extends WatchConfirmedTriggered
 
-  case class WatchTxConfirmed(replyTo: ActorRef[WatchTxConfirmedTriggered], txId: TxId, minDepth: Long) extends WatchConfirmed[WatchTxConfirmedTriggered]
+  case class RelativeDelay(parentTxId: TxId, delay: Long)
+  case class WatchTxConfirmed(replyTo: ActorRef[WatchTxConfirmedTriggered], txId: TxId, minDepth: Long, delay_opt: Option[RelativeDelay] = None) extends WatchConfirmed[WatchTxConfirmedTriggered]
   case class WatchTxConfirmedTriggered(blockHeight: BlockHeight, txIndex: Int, tx: Transaction) extends WatchConfirmedTriggered
 
   case class WatchParentTxConfirmed(replyTo: ActorRef[WatchParentTxConfirmedTriggered], txId: TxId, minDepth: Long) extends WatchConfirmed[WatchParentTxConfirmedTriggered]
@@ -167,6 +169,13 @@ object ZmqWatcher {
   private sealed trait AddWatchResult
   private case object Keep extends AddWatchResult
   private case object Ignore extends AddWatchResult
+
+  sealed trait WatchHint
+  /**
+   * In some cases we don't need to check watches every time a block is found and only need to check again after we
+   * reach a specific block height. This is for example the case for transactions with a CSV delay.
+   */
+  private case class CheckAfterBlock(blockHeight: BlockHeight) extends WatchHint
   // @formatter:on
 
   def apply(nodeParams: NodeParams, blockCount: AtomicLong, client: BitcoinCoreClient): Behavior[Command] =
@@ -178,7 +187,7 @@ object ZmqWatcher {
         timers.startSingleTimer(TickNewBlock, 1 second)
         // we start a timer in case we don't receive ZMQ block events
         timers.startSingleTimer(TickBlockTimeout, blockTimeout)
-        new ZmqWatcher(nodeParams, blockCount, client, context, timers).watching(Set.empty[GenericWatch], Map.empty[OutPoint, Set[GenericWatch]])
+        new ZmqWatcher(nodeParams, blockCount, client, context, timers).watching(Map.empty[GenericWatch, Option[WatchHint]], Map.empty[OutPoint, Set[GenericWatch]])
       }
     }
 
@@ -224,7 +233,7 @@ private class ZmqWatcher(nodeParams: NodeParams, blockHeight: AtomicLong, client
 
   private val watchdog = context.spawn(Behaviors.supervise(BlockchainWatchdog(nodeParams, 150 seconds)).onFailure(SupervisorStrategy.resume), "blockchain-watchdog")
 
-  private def watching(watches: Set[GenericWatch], watchedUtxos: Map[OutPoint, Set[GenericWatch]]): Behavior[Command] = {
+  private def watching(watches: Map[GenericWatch, Option[WatchHint]], watchedUtxos: Map[OutPoint, Set[GenericWatch]]): Behavior[Command] = {
     Behaviors.receiveMessage {
       case ProcessNewTransaction(tx) =>
         log.debug("analyzing txid={} tx={}", tx.txid, tx)
@@ -239,7 +248,7 @@ private class ZmqWatcher(nodeParams: NodeParams, blockHeight: AtomicLong, client
             case _: WatchPublished => // nothing to do
             case _: WatchConfirmed[_] => // nothing to do
           }
-        watches.collect {
+        watches.keySet.collect {
           case w: WatchPublished if w.txId == tx.txid => context.self ! TriggerEvent(w.replyTo, w, WatchPublishedTriggered(tx))
         }
         Behaviors.same
@@ -279,21 +288,32 @@ private class ZmqWatcher(nodeParams: NodeParams, blockHeight: AtomicLong, client
           case Failure(t) => GetBlockCountFailed(t)
           case Success(currentHeight) => PublishBlockHeight(currentHeight)
         }
-        // TODO: beware of the herd effect
-        KamonExt.timeFuture(Metrics.NewBlockCheckConfirmedDuration.withoutTags()) {
-          Future.sequence(watches.collect {
-            case w: WatchPublished => checkPublished(w)
-            case w: WatchConfirmed[_] => checkConfirmed(w)
-          })
-        }
         Behaviors.same
 
       case PublishBlockHeight(currentHeight) =>
         log.debug("setting blockHeight={}", currentHeight)
         blockHeight.set(currentHeight.toLong)
         context.system.eventStream ! EventStream.Publish(CurrentBlockHeight(currentHeight))
+        // TODO: should we try to mitigate the herd effect and not check all watches immediately?
+        KamonExt.timeFuture(Metrics.NewBlockCheckConfirmedDuration.withoutTags()) {
+          Future.sequence(watches.collect {
+            case (w: WatchPublished, _) => checkPublished(w)
+            case (w: WatchConfirmed[_], hint) =>
+              hint match {
+                case Some(CheckAfterBlock(delayUntilBlock)) if currentHeight < delayUntilBlock => Future.successful(())
+                case _ => checkConfirmed(w, currentHeight)
+              }
+          })
+        }
         Behaviors.same
 
+      case SetWatchHint(w, hint) =>
+        val watches1 = watches.get(w) match {
+          case Some(_) => watches + (w -> Some(hint))
+          case None => watches
+        }
+        watching(watches1, watchedUtxos)
+
       case TriggerEvent(replyTo, watch, event) =>
         if (watches.contains(watch)) {
           log.debug("triggering {}", watch)
@@ -323,7 +343,7 @@ private class ZmqWatcher(nodeParams: NodeParams, blockHeight: AtomicLong, client
             checkSpent(w)
             Keep
           case w: WatchConfirmed[_] =>
-            checkConfirmed(w)
+            checkConfirmed(w, BlockHeight(blockHeight.get()))
             Keep
           case w: WatchPublished =>
             checkPublished(w)
@@ -333,14 +353,14 @@ private class ZmqWatcher(nodeParams: NodeParams, blockHeight: AtomicLong, client
           case Keep =>
             log.debug("adding watch {}", w)
             context.watchWith(w.replyTo, StopWatching(w.replyTo))
-            watching(watches + w, addWatchedUtxos(watchedUtxos, w))
+            watching(watches + (w -> None), addWatchedUtxos(watchedUtxos, w))
           case Ignore =>
             Behaviors.same
         }
 
       case StopWatching(origin) =>
-        // we remove watches associated to dead actors
-        val deprecatedWatches = watches.filter(_.replyTo == origin)
+        // We remove watches associated to dead actors.
+        val deprecatedWatches = watches.keySet.filter(_.replyTo == origin)
         val watchedUtxos1 = deprecatedWatches.foldLeft(watchedUtxos) { case (m, w) => removeWatchedUtxos(m, w) }
         watching(watches -- deprecatedWatches, watchedUtxos1)
 
@@ -353,7 +373,7 @@ private class ZmqWatcher(nodeParams: NodeParams, blockHeight: AtomicLong, client
         Behaviors.same
 
       case r: ListWatches =>
-        r.replyTo ! watches
+        r.replyTo ! watches.keySet
         Behaviors.same
 
     }
@@ -414,7 +434,7 @@ private class ZmqWatcher(nodeParams: NodeParams, blockHeight: AtomicLong, client
     client.getTransaction(w.txId).map(tx => context.self ! TriggerEvent(w.replyTo, w, WatchPublishedTriggered(tx)))
   }
 
-  private def checkConfirmed(w: WatchConfirmed[_ <: WatchConfirmedTriggered]): Future[Unit] = {
+  private def checkConfirmed(w: WatchConfirmed[_ <: WatchConfirmedTriggered], currentHeight: BlockHeight): Future[Unit] = {
     log.debug("checking confirmations of txid={}", w.txId)
     // NB: this is very inefficient since internally we call `getrawtransaction` three times, but it doesn't really
     // matter because this only happens once, when the watched transaction has reached min_depth
@@ -431,7 +451,33 @@ private class ZmqWatcher(nodeParams: NodeParams, blockHeight: AtomicLong, client
             }
           }
         }
-      case _ => Future.successful((): Unit)
+      case Some(confirmations) =>
+        // Once the transaction is confirmed, we don't need to check again at every new block, we only need to check
+        // again once we should have reached the minimum depth to verify that there hasn't been a reorg.
+        context.self ! SetWatchHint(w, CheckAfterBlock(currentHeight + w.minDepth - confirmations))
+        Future.successful(())
+      case None =>
+        w match {
+          case WatchTxConfirmed(_, _, _, Some(relativeDelay)) =>
+            log.debug("txId={} has a relative delay of {} blocks, checking parentTxId={}", w.txId, relativeDelay.delay, relativeDelay.parentTxId)
+            // Note how we add one block to avoid an off-by-one:
+            //  - if the parent is confirmed at block P
+            //  - the CSV delay is D and the minimum depth is M
+            //  - the first block that can include the child is P + D
+            //  - the first block at which we can reach minimum depth is P + D + M
+            //  - if we are currently at block P + N, the parent has C = N + 1 confirmations
+            //  - we want to check at block P + N + D + M + 1 - C = P + N + D + M + 1 - (N + 1) = P + D + M
+            val delay = relativeDelay.delay + w.minDepth + 1
+            client.getTxConfirmations(relativeDelay.parentTxId).map(_.getOrElse(0)).collect {
+              case confirmations if confirmations < delay => context.self ! SetWatchHint(w, CheckAfterBlock(currentHeight + delay - confirmations))
+            }
+          case _ =>
+            // The transaction is unconfirmed: we don't need to check again at every new block: we can check only once
+            // every minDepth blocks, which is more efficient. If the transaction is included at the current height in
+            // a reorg, we will trigger the watch one block later than expected, but this is fine.
+            context.self ! SetWatchHint(w, CheckAfterBlock(currentHeight + w.minDepth))
+            Future.successful(())
+        }
     }
   }
 

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

@@ -115,9 +115,9 @@ object Channel {
     def spawnTxPublisher(context: ActorContext, remoteNodeId: PublicKey): typed.ActorRef[TxPublisher.Command]
   }
 
-  case class SimpleTxPublisherFactory(nodeParams: NodeParams, watcher: typed.ActorRef[ZmqWatcher.Command], bitcoinClient: BitcoinCoreClient) extends TxPublisherFactory {
+  case class SimpleTxPublisherFactory(nodeParams: NodeParams, bitcoinClient: BitcoinCoreClient) extends TxPublisherFactory {
     override def spawnTxPublisher(context: ActorContext, remoteNodeId: PublicKey): typed.ActorRef[TxPublisher.Command] = {
-      context.spawn(Behaviors.supervise(TxPublisher(nodeParams, remoteNodeId, TxPublisher.SimpleChildFactory(nodeParams, bitcoinClient, watcher))).onFailure(typed.SupervisorStrategy.restart), "tx-publisher")
+      context.spawn(Behaviors.supervise(TxPublisher(nodeParams, remoteNodeId, TxPublisher.SimpleChildFactory(nodeParams, bitcoinClient))).onFailure(typed.SupervisorStrategy.restart), "tx-publisher")
     }
   }
 
@@ -1714,7 +1714,7 @@ class Channel(val nodeParams: NodeParams, val wallet: OnChainChannelFunder with
           val (localCommitPublished1, claimHtlcTx_opt) = Closing.LocalClose.claimHtlcDelayedOutput(localCommitPublished, keyManager, d.commitments.latest, tx, nodeParams.currentFeerates, nodeParams.onChainFeeConf, d.finalScriptPubKey)
           claimHtlcTx_opt.foreach(claimHtlcTx => {
             txPublisher ! PublishFinalTx(claimHtlcTx, claimHtlcTx.fee, None)
-            blockchain ! WatchTxConfirmed(self, claimHtlcTx.tx.txid, nodeParams.channelConf.minDepthBlocks)
+            blockchain ! WatchTxConfirmed(self, claimHtlcTx.tx.txid, nodeParams.channelConf.minDepthBlocks, Some(RelativeDelay(tx.txid, d.commitments.params.remoteParams.toSelfDelay.toInt.toLong)))
           })
           Closing.updateLocalCommitPublished(localCommitPublished1, tx)
         }),