Don't rebroadcast announcements for spent channels (#3235)

* Don't rebroadcast spent channels

We introduced a mechanism to keep spent channels in the `Router` until
the spending transaction has enough confirmations to avoid removing
spliced channels from our graph too eagerly. An unwanted side-effect
is that while we're waiting for confirmations, we may broadcast channel
announcements for these spent channels, which looks like spam to our
peers.

We now filter out channels that have been spent for which we're waiting
for confirmations when sending out gossip.

* Notify front nodes immediately when channel is spent

Another side-effect of the splice change is that we didn't notify front
nodes that the channel was spent before it reached enough confirmations
so we kept sending channel gossip to our peers.

We now immediately notify the front nodes so that they stop sending this
channel to our peers.

Author: t-bast

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

@@ -161,11 +161,18 @@ class Router(val nodeParams: NodeParams, watcher: typed.ActorRef[ZmqWatcher.Comm
       stay()
 
     case Event(TickBroadcast, d) =>
-      if (d.rebroadcast.channels.isEmpty && d.rebroadcast.updates.isEmpty && d.rebroadcast.nodes.isEmpty) {
+      // We don't rebroadcast channels that have recently been spent: it may be a splice, but it makes more sense for
+      // the receiver to directly receive the post-splice announcement.
+      val spentChannels = d.spentChannels.values.flatten.toSet[ShortChannelId]
+      val rebroadcast = d.rebroadcast.copy(
+        channels = d.rebroadcast.channels.filterNot { case (c, _) => spentChannels.contains(c.shortChannelId) },
+        updates = d.rebroadcast.updates.filterNot { case (u, _) => spentChannels.contains(u.shortChannelId) }
+      )
+      if (rebroadcast.channels.isEmpty && rebroadcast.updates.isEmpty && rebroadcast.nodes.isEmpty) {
         stay()
       } else {
-        log.debug("staggered broadcast details: channels={} updates={} nodes={}", d.rebroadcast.channels.size, d.rebroadcast.updates.size, d.rebroadcast.nodes.size)
-        context.system.eventStream.publish(d.rebroadcast)
+        log.debug("staggered broadcast details: channels={} updates={} nodes={}", rebroadcast.channels.size, rebroadcast.updates.size, rebroadcast.nodes.size)
+        context.system.eventStream.publish(rebroadcast)
         stay() using d.copy(rebroadcast = Rebroadcast(channels = Map.empty, updates = Map.empty, nodes = Map.empty))
       }
 
@@ -269,6 +276,8 @@ class Router(val nodeParams: NodeParams, watcher: typed.ActorRef[ZmqWatcher.Comm
         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)
+          // We immediately notify front nodes, to ensure that we stop broadcasting this channel to our peers.
+          context.system.eventStream.publish(ChannelLost(shortChannelId))
           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
@@ -312,14 +321,18 @@ class Router(val nodeParams: NodeParams, watcher: typed.ActorRef[ZmqWatcher.Comm
       stay() using Sync.handleSendChannelQuery(d, s)
 
     case Event(PeerRoutingMessage(peerConnection, remoteNodeId, q: QueryChannelRange), d) =>
-      Sync.handleQueryChannelRange(d.channels, nodeParams.routerConf, RemoteGossip(peerConnection, remoteNodeId), q)
+      val spentChannels = d.spentChannels.values.flatten.toSet
+      val channels = d.channels.filterNot { case (scid, _) => spentChannels.contains(scid) }
+      Sync.handleQueryChannelRange(channels, nodeParams.routerConf, RemoteGossip(peerConnection, remoteNodeId), q)
       stay()
 
     case Event(PeerRoutingMessage(peerConnection, remoteNodeId, r: ReplyChannelRange), d) =>
       stay() using Sync.handleReplyChannelRange(d, nodeParams.routerConf, RemoteGossip(peerConnection, remoteNodeId), r)
 
     case Event(PeerRoutingMessage(peerConnection, remoteNodeId, q: QueryShortChannelIds), d) =>
-      Sync.handleQueryShortChannelIds(d.nodes, d.channels, RemoteGossip(peerConnection, remoteNodeId), q)
+      val spentChannels = d.spentChannels.values.flatten.toSet
+      val channels = d.channels.filterNot { case (scid, _) => spentChannels.contains(scid) }
+      Sync.handleQueryShortChannelIds(d.nodes, channels, RemoteGossip(peerConnection, remoteNodeId), q)
       stay()
 
     case Event(PeerRoutingMessage(peerConnection, remoteNodeId, r: ReplyShortChannelIdsEnd), d) =>

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

@@ -168,7 +168,7 @@ object Sync {
     var updateCount = 0
     var nodeCount = 0
 
-    processChannelQuery(nodes, channels)(
+    processChannelQuery(nodes, channels,
       q.shortChannelIds.array,
       flags,
       ca => {
@@ -291,13 +291,13 @@ object Sync {
    * @param onNode    called when a node announcement matches
    *
    */
-  def processChannelQuery(nodes: Map[PublicKey, NodeAnnouncement],
-                          channels: SortedMap[RealShortChannelId, PublicChannel])(
-                           ids: List[RealShortChannelId],
-                           flags: List[Long],
-                           onChannel: ChannelAnnouncement => Unit,
-                           onUpdate: ChannelUpdate => Unit,
-                           onNode: NodeAnnouncement => Unit)(implicit log: LoggingAdapter): Unit = {
+  private def processChannelQuery(nodes: Map[PublicKey, NodeAnnouncement],
+                                  channels: SortedMap[RealShortChannelId, PublicChannel],
+                                  ids: List[RealShortChannelId],
+                                  flags: List[Long],
+                                  onChannel: ChannelAnnouncement => Unit,
+                                  onUpdate: ChannelUpdate => Unit,
+                                  onNode: NodeAnnouncement => Unit)(implicit log: LoggingAdapter): Unit = {
     import QueryShortChannelIdsTlv.QueryFlagType
 
     // we loop over channel ids and query flag. We track node Ids for node announcement
@@ -500,7 +500,7 @@ object Sync {
       checksums = checksums)
   }
 
-  def addToSync(syncMap: Map[PublicKey, Syncing], current: Syncing, remoteNodeId: PublicKey, pending: List[QueryShortChannelIds]): (Map[PublicKey, Syncing], Option[QueryShortChannelIds]) = {
+  private def addToSync(syncMap: Map[PublicKey, Syncing], current: Syncing, remoteNodeId: PublicKey, pending: List[QueryShortChannelIds]): (Map[PublicKey, Syncing], Option[QueryShortChannelIds]) = {
     pending match {
       case head :: rest =>
         // they may send back several reply_channel_range messages for a single query_channel_range query, and we must not

eclair-core/src/test/scala/fr/acinq/eclair/router/RouterSpec.scala

@@ -303,6 +303,28 @@ class RouterSpec extends BaseRouterSpec {
       router ! Router.TickBroadcast
       eventListener.expectNoMessage(100 millis)
     }
+    {
+      // funding tx spent (splice)
+      val priv_z = randomKey()
+      val priv_funding_z = randomKey()
+      val chan_az = channelAnnouncement(RealShortChannelId(BlockHeight(420000), 0, 0), priv_z, priv_a, priv_funding_z, priv_funding_a)
+      peerConnection.send(router, PeerRoutingMessage(peerConnection.ref, remoteNodeId, chan_az))
+      assert(watcher.expectMsgType[ValidateRequest].ann == chan_az)
+      watcher.send(router, ValidateResult(chan_az, Right(Transaction(2, Nil, TxOut(1000000 sat, write(pay2wsh(Scripts.multiSig2of2(funding_a, priv_funding_z.publicKey)))) :: Nil, 0), UtxoStatus.Unspent)))
+      peerConnection.expectMsg(TransportHandler.ReadAck(chan_az))
+      peerConnection.expectMsg(GossipDecision.Accepted(chan_az))
+      assert(watcher.expectMsgType[WatchExternalChannelSpent].shortChannelId == chan_az.shortChannelId)
+      eventListener.expectMsg(ChannelsDiscovered(SingleChannelDiscovered(chan_az, 1000000 sat, None, None) :: Nil))
+      peerConnection.expectNoMessage(100 millis)
+      eventListener.expectNoMessage(100 millis)
+      // The channel is spent by a splice transaction: it is removed from the rebroadcast list.
+      router ! WatchExternalChannelSpentTriggered(RealShortChannelId(BlockHeight(420000), 0, 0), Some(spendingTx(funding_a, priv_funding_z.publicKey)))
+      watcher.expectMsgType[WatchTxConfirmed]
+      // We notify front nodes to ensure they also stop broadcasting this channel.
+      eventListener.expectMsg(ChannelLost(RealShortChannelId(BlockHeight(420000), 0, 0)))
+      router ! Router.TickBroadcast
+      eventListener.expectNoMessage(100 millis)
+    }
 
     watcher.expectNoMessage(100 millis)
   }
@@ -343,6 +365,7 @@ class RouterSpec extends BaseRouterSpec {
 
     router ! WatchExternalChannelSpentTriggered(scid_ab, Some(spendingTx(funding_a, funding_b)))
     watcher.expectMsgType[WatchTxConfirmed]
+    eventListener.expectMsg(ChannelLost(scid_ab))
     router ! WatchTxConfirmedTriggered(BlockHeight(0), 0, spendingTx(funding_a, funding_b))
     watcher.expectMsg(UnwatchExternalChannelSpent(channels(scid_ab).fundingTxId, ShortChannelId.outputIndex(scid_ab)))
     eventListener.expectMsg(ChannelLost(scid_ab))
@@ -365,6 +388,7 @@ class RouterSpec extends BaseRouterSpec {
 
     router ! WatchExternalChannelSpentTriggered(scid_bc, Some(spendingTx(funding_b, funding_c)))
     watcher.expectMsgType[WatchTxConfirmed]
+    eventListener.expectMsg(ChannelLost(scid_bc))
     router ! WatchTxConfirmedTriggered(BlockHeight(0), 0, spendingTx(funding_b, funding_c))
     watcher.expectMsg(UnwatchExternalChannelSpent(channels(scid_bc).fundingTxId, ShortChannelId.outputIndex(scid_bc)))
     eventListener.expectMsg(ChannelLost(scid_bc))
@@ -406,6 +430,7 @@ class RouterSpec extends BaseRouterSpec {
     // The channel is closed, now we can remove it from the DB.
     router ! WatchExternalChannelSpentTriggered(scid_au, Some(spendingTx(funding_a, priv_funding_u.publicKey)))
     assert(watcher.expectMsgType[WatchTxConfirmed].txId == spendingTx(funding_a, priv_funding_u.publicKey).txid)
+    eventListener.expectMsg(ChannelLost(scid_au))
     router ! WatchTxConfirmedTriggered(BlockHeight(0), 0, spendingTx(funding_a, priv_funding_u.publicKey))
     watcher.expectMsg(UnwatchExternalChannelSpent(channels(scid_au).fundingTxId, ShortChannelId.outputIndex(scid_au)))
     eventListener.expectMsg(ChannelLost(scid_au))
@@ -1196,6 +1221,7 @@ class RouterSpec extends BaseRouterSpec {
       assert(w.txId == spliceTx1.txid)
       assert(w.minDepth == 12)
     }
+    eventListener.expectMsg(ChannelLost(scid_ab))
     eventListener.expectNoMessage(100 millis)
 
     // Channel ab is spent and confirmed by an RBF of splice tx.
@@ -1206,6 +1232,7 @@ class RouterSpec extends BaseRouterSpec {
       assert(w.txId == spliceTx2.txid)
       assert(w.minDepth == 12)
     }
+    eventListener.expectMsg(ChannelLost(scid_ab))
     eventListener.expectNoMessage(100 millis)
 
     // The splice of channel ab is announced.
@@ -1246,37 +1273,43 @@ class RouterSpec extends BaseRouterSpec {
       assert(w.txId == spliceTx_ab.txid)
       assert(w.minDepth == 12)
     }
+    eventListener.expectMsg(ChannelLost(scid_ab))
 
     // Channel bc is spent by a splice tx.
     router ! WatchExternalChannelSpentTriggered(scid_bc, Some(spliceTx_bc))
     inside(watcher.expectMsgType[WatchTxConfirmed]) { w =>
       assert(w.txId == spliceTx_bc.txid)
       assert(w.minDepth == 12)
     }
+    eventListener.expectMsg(ChannelLost(scid_bc))
 
     // Channel bc2 is spent by a splice tx.
     router ! WatchExternalChannelSpentTriggered(scid_bc2, Some(spliceTx_bc2))
     inside(watcher.expectMsgType[WatchTxConfirmed]) { w =>
       assert(w.txId == spliceTx_bc2.txid)
       assert(w.minDepth == 12)
     }
+    eventListener.expectMsg(ChannelLost(scid_bc2))
 
     // Channels ab, bc and bc2 are all spent by the same batch splice tx.
     router ! WatchExternalChannelSpentTriggered(scid_ab, Some(batchSpliceTx))
     inside(watcher.expectMsgType[WatchTxConfirmed]) { w =>
       assert(w.txId == batchSpliceTx.txid)
       assert(w.minDepth == 12)
     }
+    eventListener.expectMsg(ChannelLost(scid_ab))
     router ! WatchExternalChannelSpentTriggered(scid_bc, Some(batchSpliceTx))
     inside(watcher.expectMsgType[WatchTxConfirmed]) { w =>
       assert(w.txId == batchSpliceTx.txid)
       assert(w.minDepth == 12)
     }
+    eventListener.expectMsg(ChannelLost(scid_bc))
     router ! WatchExternalChannelSpentTriggered(scid_bc2, Some(batchSpliceTx))
     inside(watcher.expectMsgType[WatchTxConfirmed]) { w =>
       assert(w.txId == batchSpliceTx.txid)
       assert(w.minDepth == 12)
     }
+    eventListener.expectMsg(ChannelLost(scid_bc2))
 
     // Channels ab, bc and bc2 are also all spent by an RBF of the batch splice tx.
     val newCapacity_ab_RBF = newCapacity_ab - 1000.sat
@@ -1286,16 +1319,19 @@ class RouterSpec extends BaseRouterSpec {
       assert(w.txId == batchSpliceTx_RBF.txid)
       assert(w.minDepth == 12)
     }
+    eventListener.expectMsg(ChannelLost(scid_ab))
     router ! WatchExternalChannelSpentTriggered(scid_bc, Some(batchSpliceTx_RBF))
     inside(watcher.expectMsgType[WatchTxConfirmed]) { w =>
       assert(w.txId == batchSpliceTx_RBF.txid)
       assert(w.minDepth == 12)
     }
+    eventListener.expectMsg(ChannelLost(scid_bc))
     router ! WatchExternalChannelSpentTriggered(scid_bc2, Some(batchSpliceTx_RBF))
     inside(watcher.expectMsgType[WatchTxConfirmed]) { w =>
       assert(w.txId == batchSpliceTx_RBF.txid)
       assert(w.minDepth == 12)
     }
+    eventListener.expectMsg(ChannelLost(scid_bc2))
 
     // The router tracks the possible spending txs for channels ab, bc and bc2.
     val sender = TestProbe()