Don't scan the blockchain for spent external channels (#3226)

When an external channel is spent, we don't immediately remove it from
our network graph in case the spending transaction is a splice (see
lightning/bolts#1270 for more details).

A side-effect of this change, introduced in #2936, is that when we
start watching a channel after receiving its `channel_announcement`,
we will scan the blockchain if it is actually already spent. This can
be expensive if peers send us `channel_announcement`s for channels
that have been spent a long time ago since `bitcoind` doesn't provide
an index for spending transactions. It is also misleading, because if
we give up after scanning X blocks of the blockchain, we will create
a log line saying that funds are at risk: they're never at risk since
those are not our channels.

This commit fixes this issue by only checking whether the channel is
already spent by a confirmed transaction or not when setting the watch
(which is an inexpensive and efficient RPC call to `bitcoind`), without
scanning the blockchain to find the spending transaction. If it is
already spent, we immediately remove it from our network graph, even
if the spending transaction was actually a splice. This is fine, since
that channel will be re-added to our graph whenever we receive the
`channel_announcement` for the splice. In the worst case, we will simply
not route through an actually available channels for a few blocks while
its splice transaction is confirming.

Co-authored-by: pm47 <pm.padiou@gmail.com>

Author: t-bast

eclair-core/src/main/scala/fr/acinq/eclair/blockchain/Monitoring.scala

@@ -24,6 +24,8 @@ object Monitoring {
 
   object Metrics {
     val NewBlockCheckConfirmedDuration: Metric.Timer = Kamon.timer("bitcoin.watcher.newblock.checkconfirmed")
+    val Watches: Metric.Gauge = Kamon.gauge("bitcoin.watcher.watches")
+    val WatchedUtxos: Metric.Gauge = Kamon.gauge("bitcoin.watcher.utxos")
     val RpcBasicInvokeCount: Metric.Counter = Kamon.counter("bitcoin.rpc.basic.invoke.count")
     val RpcBasicInvokeDuration: Metric.Timer = Kamon.timer("bitcoin.rpc.basic.invoke.duration")
     val RpcBatchInvokeDuration: Metric.Timer = Kamon.timer("bitcoin.rpc.batch.invoke.duration")
@@ -45,6 +47,7 @@ object Monitoring {
     val Wallet = "wallet"
     val Priority = "priority"
     val Provider = "provider"
+    val WatchType = "watch-type"
 
     object Priorities {
       val Minimum = "0-minimum"

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

@@ -121,13 +121,10 @@ object ZmqWatcher {
   }
 
   /** This event is sent when a [[WatchSpent]] condition is met. */
-  sealed trait WatchSpentTriggered extends WatchTriggered {
-    /** Transaction spending the watched outpoint. */
-    def spendingTx: Transaction
-  }
+  sealed trait WatchSpentTriggered extends WatchTriggered
 
   case class WatchExternalChannelSpent(replyTo: ActorRef[WatchExternalChannelSpentTriggered], txId: TxId, outputIndex: Int, shortChannelId: RealShortChannelId) extends WatchSpent[WatchExternalChannelSpentTriggered] { override def hints: Set[TxId] = Set.empty }
-  case class WatchExternalChannelSpentTriggered(shortChannelId: RealShortChannelId, spendingTx: Transaction) extends WatchSpentTriggered
+  case class WatchExternalChannelSpentTriggered(shortChannelId: RealShortChannelId, spendingTx_opt: Option[Transaction]) extends WatchSpentTriggered
   case class UnwatchExternalChannelSpent(txId: TxId, outputIndex: Int) extends Command
 
   case class WatchFundingSpent(replyTo: ActorRef[WatchFundingSpentTriggered], txId: TxId, outputIndex: Int, hints: Set[TxId]) extends WatchSpent[WatchFundingSpentTriggered]
@@ -231,7 +228,7 @@ private class ZmqWatcher(nodeParams: NodeParams, blockHeight: AtomicLong, client
           .flatMap(watchedUtxos.get)
           .flatten
           .foreach {
-            case w: WatchExternalChannelSpent => context.self ! TriggerEvent(w.replyTo, w, WatchExternalChannelSpentTriggered(w.shortChannelId, tx))
+            case w: WatchExternalChannelSpent => context.self ! TriggerEvent(w.replyTo, w, WatchExternalChannelSpentTriggered(w.shortChannelId, Some(tx)))
             case w: WatchFundingSpent => context.self ! TriggerEvent(w.replyTo, w, WatchFundingSpentTriggered(tx))
             case w: WatchOutputSpent => context.self ! TriggerEvent(w.replyTo, w, WatchOutputSpentTriggered(w.amount, tx))
             case _: WatchPublished => // nothing to do
@@ -322,12 +319,14 @@ private class ZmqWatcher(nodeParams: NodeParams, blockHeight: AtomicLong, client
         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?
-        val watchExternalChannelCount = watches.keySet.count(_.isInstanceOf[WatchExternalChannelSpent])
-        val watchFundingSpentCount = watches.keySet.count(_.isInstanceOf[WatchFundingSpent])
-        val watchOutputSpentCount = watches.keySet.count(_.isInstanceOf[WatchOutputSpent])
-        val watchPublishedCount = watches.keySet.count(_.isInstanceOf[WatchPublished])
-        val watchConfirmedCount = watches.keySet.count(_.isInstanceOf[WatchConfirmed[_]])
-        log.info("{} watched utxos: external-channels={}, funding-spent={}, output-spent={}, tx-published={}, tx-confirmed={}", watchedUtxos.size, watchExternalChannelCount, watchFundingSpentCount, watchOutputSpentCount, watchPublishedCount, watchConfirmedCount)
+        Metrics.WatchedUtxos.withoutTags().update(watchedUtxos.size)
+        log.info("currently watching {} utxos with {} watches", watchedUtxos.size, watches.size)
+        watches.keys
+          .groupBy(_.getClass.getSimpleName)
+          .foreach { case (t, list) =>
+            Metrics.Watches.withTag(Monitoring.Tags.WatchType, t).update(list.size)
+            log.info("we have {} {} currently registered", list.size, t)
+          }
         KamonExt.timeFuture(Metrics.NewBlockCheckConfirmedDuration.withoutTags()) {
           Future.sequence(watches.collect {
             case (w: WatchPublished, _) => checkPublished(w)
@@ -429,41 +428,54 @@ private class ZmqWatcher(nodeParams: NodeParams, blockHeight: AtomicLong, client
   }
 
   private def checkSpent(w: WatchSpent[_ <: WatchSpentTriggered]): Future[Unit] = {
-    // first let's see if the parent tx was published or not
+    // First let's see if the parent tx was published or not before checking whether it has been spent.
     client.getTxConfirmations(w.txId).collect {
-      case Some(_) =>
-        // parent tx was published, we need to make sure this particular output has not been spent
-        client.isTransactionOutputSpendable(w.txId, w.outputIndex, includeMempool = true).collect {
-          case false =>
-            // the output has been spent, let's find the spending tx
-            // if we know some potential spending txs, we try to fetch them directly
-            Future.sequence(w.hints.map(txid => client.getTransaction(txid).map(Some(_)).recover { case _ => None }))
-              .map(_.flatten) // filter out errors and hint transactions that can't be found
-              .map(hintTxs => {
-                hintTxs.find(tx => tx.txIn.exists(i => i.outPoint.txid == w.txId && i.outPoint.index == w.outputIndex)) match {
-                  case Some(spendingTx) =>
-                    log.info(s"${w.txId}:${w.outputIndex} has already been spent by a tx provided in hints: txid=${spendingTx.txid}")
-                    context.self ! ProcessNewTransaction(spendingTx)
-                  case None =>
-                    // The hints didn't help us, let's search for the spending transaction.
-                    log.info(s"${w.txId}:${w.outputIndex} has already been spent, looking for the spending tx in the mempool")
-                    client.lookForMempoolSpendingTx(w.txId, w.outputIndex).map(Some(_)).recover { case _ => None }.map {
-                      case Some(spendingTx) =>
-                        log.info(s"found tx spending ${w.txId}:${w.outputIndex} in the mempool: txid=${spendingTx.txid}")
-                        context.self ! ProcessNewTransaction(spendingTx)
-                      case None =>
-                        // no luck, we have to do it the hard way...
-                        log.warn(s"${w.txId}:${w.outputIndex} has already been spent, spending tx not in the mempool, looking in the blockchain...")
-                        client.lookForSpendingTx(None, w.txId, w.outputIndex, nodeParams.channelConf.maxChannelSpentRescanBlocks).map { spendingTx =>
-                          log.warn(s"found the spending tx of ${w.txId}:${w.outputIndex} in the blockchain: txid=${spendingTx.txid}")
+      case Some(_) => w match {
+        case w: WatchExternalChannelSpent =>
+          // This is an external channels: funds are not at risk, so we don't need to scan the blockchain to find the
+          // spending transaction, it is costly and unnecessary. We simply check whether the output has already been
+          // spent by a confirmed transaction.
+          client.isTransactionOutputSpent(w.txId, w.outputIndex).collect {
+            case true =>
+              // The output has been spent, so we trigger the watch without including the spending transaction.
+              context.self ! TriggerEvent(w.replyTo, w, WatchExternalChannelSpentTriggered(w.shortChannelId, None))
+          }
+        case _ =>
+          // The parent tx was published, we need to make sure this particular output has not been spent.
+          client.isTransactionOutputSpendable(w.txId, w.outputIndex, includeMempool = true).collect {
+            case false =>
+              // The output has been spent, let's find the spending tx.
+              // If we know some potential spending txs, we try to fetch them directly.
+              Future.sequence(w.hints.map(txid => client.getTransaction(txid).map(Some(_)).recover { case _ => None }))
+                .map(_.flatten) // filter out errors and hint transactions that can't be found
+                .map(hintTxs => {
+                  hintTxs.find(tx => tx.txIn.exists(i => i.outPoint.txid == w.txId && i.outPoint.index == w.outputIndex)) match {
+                    case Some(spendingTx) =>
+                      log.info("{}:{} has already been spent by a tx provided in hints: txid={}", w.txId, w.outputIndex, spendingTx.txid)
+                      context.self ! ProcessNewTransaction(spendingTx)
+                    case None =>
+                      // The hints didn't help us, let's search for the spending transaction in the mempool.
+                      log.info("{}:{} has already been spent, looking for the spending tx in the mempool", w.txId, w.outputIndex)
+                      client.lookForMempoolSpendingTx(w.txId, w.outputIndex).map(Some(_)).recover { case _ => None }.map {
+                        case Some(spendingTx) =>
+                          log.info("found tx spending {}:{} in the mempool: txid={}", w.txId, w.outputIndex, spendingTx.txid)
                           context.self ! ProcessNewTransaction(spendingTx)
-                        }.recover {
-                          case _ => log.warn(s"could not find the spending tx of ${w.txId}:${w.outputIndex} in the blockchain, funds are at risk")
-                        }
-                    }
-                }
-              })
-        }
+                        case None =>
+                          // The spending transaction isn't in the mempool, so it must be a transaction that confirmed
+                          // before we set the watch. We have to scan the blockchain to find it, which is expensive
+                          // since bitcoind doesn't provide indexes for this scenario.
+                          log.warn("{}:{} has already been spent, spending tx not in the mempool, looking in the blockchain...", w.txId, w.outputIndex)
+                          client.lookForSpendingTx(None, w.txId, w.outputIndex, nodeParams.channelConf.maxChannelSpentRescanBlocks).map { spendingTx =>
+                            log.warn("found the spending tx of {}:{} in the blockchain: txid={}", w.txId, w.outputIndex, spendingTx.txid)
+                            context.self ! ProcessNewTransaction(spendingTx)
+                          }.recover {
+                            case _ => log.warn("could not find the spending tx of {}:{} in the blockchain, funds are at risk", w.txId, w.outputIndex)
+                          }
+                      }
+                  }
+                })
+          }
+      }
     }
   }
 

eclair-core/src/main/scala/fr/acinq/eclair/router/Router.scala

@@ -263,15 +263,24 @@ class Router(val nodeParams: NodeParams, watcher: typed.ActorRef[ZmqWatcher.Comm
     case Event(r: ValidateResult, d) =>
       stay() using Validation.handleChannelValidationResponse(d, nodeParams, watcher, r)
 
-    case Event(WatchExternalChannelSpentTriggered(shortChannelId, spendingTx), d) if d.channels.contains(shortChannelId) || d.prunedChannels.contains(shortChannelId) =>
+    case Event(WatchExternalChannelSpentTriggered(shortChannelId, spendingTx_opt), d) if d.channels.contains(shortChannelId) || d.prunedChannels.contains(shortChannelId) =>
       val fundingTxId = d.channels.get(shortChannelId).orElse(d.prunedChannels.get(shortChannelId)).get.fundingTxId
-      log.info("funding tx txId={} of channelId={} has been spent by txId={}: waiting for the spending tx to have enough confirmations before removing the channel from the graph", fundingTxId, shortChannelId, spendingTx.txid)
-      watcher ! WatchTxConfirmed(self, spendingTx.txid, nodeParams.routerConf.channelSpentSpliceDelay)
-      stay() using d.copy(spentChannels = d.spentChannels.updated(spendingTx.txid, d.spentChannels.getOrElse(spendingTx.txid, Set.empty) + shortChannelId))
+      spendingTx_opt match {
+        case Some(spendingTx) =>
+          log.info("funding tx txId={} of channelId={} has been spent by txId={}: waiting for the spending tx to have enough confirmations before removing the channel from the graph", fundingTxId, shortChannelId, spendingTx.txid)
+          watcher ! WatchTxConfirmed(self, spendingTx.txid, nodeParams.routerConf.channelSpentSpliceDelay)
+          stay() using d.copy(spentChannels = d.spentChannels.updated(spendingTx.txid, d.spentChannels.getOrElse(spendingTx.txid, Set.empty) + shortChannelId))
+        case None =>
+          // If the channel was spent by a transaction that is already confirmed, it would be very inefficient to scan
+          // the blockchain for the spending transaction (which could have confirmed a long time ago), just to forget
+          // that channel. We skip scanning the blockchain and immediately forget the channel.
+          log.info("funding tx txId={} of channelId={} has already been spent by a confirmed transaction: removing the channel from the graph immediately", fundingTxId, shortChannelId)
+          stay() using Validation.handleChannelSpent(d, watcher, nodeParams.db.network, None, Set(shortChannelId))
+      }
 
     case Event(WatchTxConfirmedTriggered(_, _, spendingTx), d) =>
       d.spentChannels.get(spendingTx.txid) match {
-        case Some(shortChannelIds) => stay() using Validation.handleChannelSpent(d, watcher, nodeParams.db.network, spendingTx.txid, shortChannelIds)
+        case Some(shortChannelIds) => stay() using Validation.handleChannelSpent(d, watcher, nodeParams.db.network, Some(spendingTx.txid), shortChannelIds)
         case None => stay()
       }
 

eclair-core/src/main/scala/fr/acinq/eclair/router/Validation.scala

@@ -268,7 +268,7 @@ object Validation {
     } else d1
   }
 
-  def handleChannelSpent(d: Data, watcher: typed.ActorRef[ZmqWatcher.Command], db: NetworkDb, spendingTxId: TxId, shortChannelIds: Set[RealShortChannelId])(implicit ctx: ActorContext, log: LoggingAdapter): Data = {
+  def handleChannelSpent(d: Data, watcher: typed.ActorRef[ZmqWatcher.Command], db: NetworkDb, spendingTxId_opt: Option[TxId], shortChannelIds: Set[RealShortChannelId])(implicit ctx: ActorContext, log: LoggingAdapter): Data = {
     implicit val sender: ActorRef = ctx.self // necessary to preserve origin when sending messages to other actors
     val lostChannels = shortChannelIds.flatMap(shortChannelId => d.channels.get(shortChannelId).orElse(d.prunedChannels.get(shortChannelId)))
     log.info("funding tx for channelIds={} was spent", shortChannelIds.mkString(","))
@@ -277,7 +277,7 @@ object Validation {
     val prunedChannels1 = d.prunedChannels -- shortChannelIds
     val lostNodes = lostChannels.flatMap(lostChannel => Seq(lostChannel.nodeId1, lostChannel.nodeId2).filterNot(nodeId => hasChannels(nodeId, channels1.values)))
     // let's clean the db and send the events
-      log.info("pruning shortChannelIds={} (spent)", shortChannelIds.mkString(","))
+    log.info("pruning shortChannelIds={} (spent)", shortChannelIds.mkString(","))
     shortChannelIds.foreach(db.removeChannel(_)) // NB: this also removes channel updates
     // we also need to remove updates from the graph
     val graphWithBalances1 = lostChannels.foldLeft(d.graphWithBalances) { (graph, lostChannel) =>
@@ -298,13 +298,12 @@ object Validation {
         // we will re-add a spliced channel as a new channel later when we receive the announcement
         watcher ! UnwatchExternalChannelSpent(lostChannel.fundingTxId, outputIndex(lostChannel.ann.shortChannelId))
     }
-
     // We may have received RBF candidates for this splice: we can find them by looking at transactions that spend one
     // of the channels we're removing (note that they may spend a slightly different set of channels).
     // Those transactions cannot confirm anymore (they have been double-spent by the current one), so we should stop
     // watching them.
     val spendingTxs = d.spentChannels.filter(_._2.intersect(shortChannelIds).nonEmpty).keySet
-    (spendingTxs - spendingTxId).foreach(txId => watcher ! UnwatchTxConfirmed(txId))
+    (spendingTxs -- spendingTxId_opt.toSet).foreach(txId => watcher ! UnwatchTxConfirmed(txId))
     val spentChannels1 = d.spentChannels -- spendingTxs
     d.copy(nodes = d.nodes -- lostNodes, channels = channels1, prunedChannels = prunedChannels1, graphWithBalances = graphWithBalances1, spentChannels = spentChannels1)
   }

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

@@ -264,15 +264,15 @@ class ZmqWatcherSpec extends TestKitBaseClass with AnyFunSuiteLike with Bitcoind
       bitcoinClient.publishTransaction(tx1)
       // tx and tx1 aren't confirmed yet, but we trigger the WatchSpentTriggered event when we see tx1 in the mempool.
       probe.expectMsgAllOf(
-        WatchExternalChannelSpentTriggered(RealShortChannelId(5), tx1),
+        WatchExternalChannelSpentTriggered(RealShortChannelId(5), Some(tx1)),
         WatchFundingSpentTriggered(tx1)
       )
       // Let's confirm tx and tx1: seeing tx1 in a block should trigger both WatchSpentTriggered events again.
       bitcoinClient.getBlockHeight().pipeTo(probe.ref)
       val initialBlockHeight = probe.expectMsgType[BlockHeight]
       generateBlocks(1)
       probe.expectMsgAllOf(
-        WatchExternalChannelSpentTriggered(RealShortChannelId(5), tx1),
+        WatchExternalChannelSpentTriggered(RealShortChannelId(5), Some(tx1)),
         WatchFundingSpentTriggered(tx1)
       )
       probe.expectNoMessage(100 millis)
@@ -311,8 +311,10 @@ class ZmqWatcherSpec extends TestKitBaseClass with AnyFunSuiteLike with Bitcoind
       probe.fishForMessage() { case m: WatchOutputSpentTriggered => m.spendingTx.txid == tx1.txid }
       watcher ! StopWatching(probe.ref)
 
+      // If we watch after being spent by a confirmed transaction, we immediately trigger the watch without fetching
+      // the spending transaction.
       watcher ! WatchExternalChannelSpent(probe.ref, tx1.txid, 0, RealShortChannelId(1))
-      probe.expectMsg(WatchExternalChannelSpentTriggered(RealShortChannelId(1), tx2))
+      probe.expectMsg(WatchExternalChannelSpentTriggered(RealShortChannelId(1), None))
       watcher ! StopWatching(probe.ref)
       watcher ! WatchFundingSpent(probe.ref, tx1.txid, 0, Set.empty)
       probe.expectMsg(WatchFundingSpentTriggered(tx2))
@@ -340,7 +342,7 @@ class ZmqWatcherSpec extends TestKitBaseClass with AnyFunSuiteLike with Bitcoind
       // When publishing the transaction, the watch triggers immediately.
       bitcoinClient.publishTransaction(spendingTx1).pipeTo(sender.ref)
       sender.expectMsg(spendingTx1.txid)
-      probe.expectMsg(WatchExternalChannelSpentTriggered(RealShortChannelId(3), spendingTx1))
+      probe.expectMsg(WatchExternalChannelSpentTriggered(RealShortChannelId(3), Some(spendingTx1)))
       probe.expectNoMessage(100 millis)
 
       // If we unwatch the transaction, we will ignore when it's published.
@@ -351,7 +353,7 @@ class ZmqWatcherSpec extends TestKitBaseClass with AnyFunSuiteLike with Bitcoind
 
       // If we watch again, this will trigger immediately because the transaction is in the mempool.
       watcher ! WatchExternalChannelSpent(probe.ref, tx2.txid, outputIndex2, RealShortChannelId(5))
-      probe.expectMsg(WatchExternalChannelSpentTriggered(RealShortChannelId(5), spendingTx2))
+      probe.expectMsg(WatchExternalChannelSpentTriggered(RealShortChannelId(5), Some(spendingTx2)))
       probe.expectNoMessage(100 millis)
 
       // We make the transactions confirm while we're not watching.
@@ -365,7 +367,7 @@ class ZmqWatcherSpec extends TestKitBaseClass with AnyFunSuiteLike with Bitcoind
 
       // If we watch again after confirmation, the watch instantly triggers.
       watcher ! WatchExternalChannelSpent(probe.ref, tx1.txid, outputIndex1, RealShortChannelId(3))
-      probe.expectMsg(WatchExternalChannelSpentTriggered(RealShortChannelId(3), spendingTx1))
+      probe.expectMsg(WatchExternalChannelSpentTriggered(RealShortChannelId(3), None))
       probe.expectNoMessage(100 millis)
     })
   }