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

diff --git a/eclair-core/src/main/scala/fr/acinq/eclair/router/Router.scala b/eclair-core/src/main/scala/fr/acinq/eclair/router/Router.scala
index 2609dbd..e7a47b1 100644
--- a/eclair-core/src/main/scala/fr/acinq/eclair/router/Router.scala
+++ b/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) =>
diff --git a/eclair-core/src/main/scala/fr/acinq/eclair/router/Sync.scala b/eclair-core/src/main/scala/fr/acinq/eclair/router/Sync.scala
index ee7f64d..fab8001 100644
--- a/eclair-core/src/main/scala/fr/acinq/eclair/router/Sync.scala
+++ b/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
diff --git a/eclair-core/src/test/scala/fr/acinq/eclair/router/RouterSpec.scala b/eclair-core/src/test/scala/fr/acinq/eclair/router/RouterSpec.scala
index 6eed79d..e0540ef 100644
--- a/eclair-core/src/test/scala/fr/acinq/eclair/router/RouterSpec.scala
+++ b/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,6 +1273,7 @@ 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))
@@ -1253,6 +1281,7 @@ class RouterSpec extends BaseRouterSpec {
       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))
@@ -1260,6 +1289,7 @@ class RouterSpec extends BaseRouterSpec {
       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))
@@ -1267,16 +1297,19 @@ class RouterSpec extends BaseRouterSpec {
       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()