Channel range queries: send back node announcements

eclair-core/src/main/resources/reference.conf

@@ -116,6 +116,7 @@ eclair {
     channel-exclude-duration = 60 seconds // when a temporary channel failure is returned, we exclude the channel from our payment routes for this duration
     broadcast-interval = 60 seconds // see BOLT #7
     init-timeout = 5 minutes
+    request-node-announcements = true // if true we will ask for node annnouncements when we receive channel ids that we don't know
 
     // the values below will be used to perform route searching
     path-finding {

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

@@ -235,6 +235,7 @@ object NodeParams {
         channelExcludeDuration = FiniteDuration(config.getDuration("router.channel-exclude-duration").getSeconds, TimeUnit.SECONDS),
         routerBroadcastInterval = FiniteDuration(config.getDuration("router.broadcast-interval").getSeconds, TimeUnit.SECONDS),
         randomizeRouteSelection = config.getBoolean("router.randomize-route-selection"),
+        requestNodeAnnouncements = config.getBoolean("router.request-node-announcements"),
         searchMaxRouteLength = config.getInt("router.path-finding.max-route-length"),
         searchMaxCltv = config.getInt("router.path-finding.max-cltv"),
         searchMaxFeeBase = Satoshi(config.getLong("router.path-finding.fee-threshold-sat")),

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

@@ -50,6 +50,7 @@ import scala.util.{Random, Try}
 case class RouterConf(randomizeRouteSelection: Boolean,
                       channelExcludeDuration: FiniteDuration,
                       routerBroadcastInterval: FiniteDuration,
+                      requestNodeAnnouncements: Boolean,
                       searchMaxFeeBase: Satoshi,
                       searchMaxFeePct: Double,
                       searchMaxRouteLength: Int,
@@ -535,7 +536,7 @@ class Router(val nodeParams: NodeParams, watcher: ActorRef, initialized: Option[
         ids match {
           case Nil => acc.reverse
           case head :: tail =>
-            val flag = computeFlag(d.channels, d.updates)(head, timestamps.headOption, checksums.headOption)
+            val flag = computeFlag(d.channels, d.updates)(head, timestamps.headOption, checksums.headOption, nodeParams.routerConf.requestNodeAnnouncements)
             // 0 means nothing to query, just don't include it
             val acc1 = if (flag != 0) ShortChannelIdAndFlag(head, flag) :: acc else acc
             loop(tail, timestamps.drop(1), checksums.drop(1), acc1)
@@ -549,7 +550,7 @@ class Router(val nodeParams: NodeParams, watcher: ActorRef, initialized: Option[
 
       val (channelCount, updatesCount) = shortChannelIdAndFlags.foldLeft((0, 0)) {
         case ((c, u), ShortChannelIdAndFlag(_, flag)) =>
-          val c1 = c + (if (QueryShortChannelIdsTlv.QueryFlagType.includeAnnouncement(flag)) 1 else 0)
+          val c1 = c + (if (QueryShortChannelIdsTlv.QueryFlagType.includeChannelAnnouncement(flag)) 1 else 0)
           val u1 = u + (if (QueryShortChannelIdsTlv.QueryFlagType.includeUpdate1(flag)) 1 else 0) + (if (QueryShortChannelIdsTlv.QueryFlagType.includeUpdate2(flag)) 1 else 0)
           (c1, u1)
       }
@@ -573,26 +574,58 @@ class Router(val nodeParams: NodeParams, watcher: ActorRef, initialized: Option[
 
     case Event(PeerRoutingMessage(transport, _, routingMessage@QueryShortChannelIds(chainHash, shortChannelIds, queryFlags_opt)), d) =>
       sender ! TransportHandler.ReadAck(routingMessage)
-      val (channelCount, updatesCount) = shortChannelIds.array
-        .zipWithIndex
-        .foldLeft((0, 0)) {
-          case ((c, u), (shortChannelId, idx)) =>
-            var c1 = c
-            var u1 = u
-            val flag = routingMessage.queryFlags_opt.map(_.array(idx)).getOrElse(QueryShortChannelIdsTlv.QueryFlagType.INCLUDE_ALL)
-            d.channels.get(shortChannelId) match {
-              case None => log.warning("received query for shortChannelId={} that we don't have", shortChannelId)
-              case Some(ca) =>
-                if (QueryShortChannelIdsTlv.QueryFlagType.includeAnnouncement(flag)) {
-                  transport ! ca
-                  c1 = c1 + 1
-                }
-                if (QueryShortChannelIdsTlv.QueryFlagType.includeUpdate1(flag)) d.updates.get(ChannelDesc(ca.shortChannelId, ca.nodeId1, ca.nodeId2)).foreach { u => transport ! u; u1 = u1 + 1 }
-                if (QueryShortChannelIdsTlv.QueryFlagType.includeUpdate2(flag)) d.updates.get(ChannelDesc(ca.shortChannelId, ca.nodeId2, ca.nodeId1)).foreach { u => transport ! u; u1 = u1 + 1 }
+      val flags = routingMessage.queryFlags_opt.map(_.array).getOrElse(List.empty[Long])
+
+      // we loop over channel ids and query flag and send what the sender requested
+      // we keep track of how many announcements and updates we've sent, and we also track node Ids for node announcement
+      // we've already sent to avoid sending them multiple times, as requested by the BOLTs
+      @tailrec
+      def loop(ids: List[ShortChannelId], flags: List[Long], numca: Int = 0, numcu: Int = 0, sent: Set[PublicKey] = Set.empty[PublicKey]): (Int, Int, Int) = ids match {
+        case Nil => (numca, numcu, sent.size)
+        case head :: tail if !d.channels.contains(head) =>
+          log.warning("received query for shortChannelId={} that we don't have", head)
+          loop(tail, flags.drop(1), numca, numcu, sent)
+        case head :: tail =>
+          var numca1 = numca
+          var numcu1 = numcu
+          var sent1 = sent
+          val ca = d.channels(head)
+          val flag_opt = flags.headOption
+          // no flag means send everything
+
+          if (flag_opt.map(QueryShortChannelIdsTlv.QueryFlagType.includeChannelAnnouncement).getOrElse(true)) {
+            transport ! ca
+            numca1 = numca1 + 1
+          }
+          if (flag_opt.map(QueryShortChannelIdsTlv.QueryFlagType.includeUpdate1).getOrElse(true)) {
+            d.updates.get(ChannelDesc(ca.shortChannelId, ca.nodeId1, ca.nodeId2)).foreach { u =>
+              transport ! u
+              numcu1 = numcu1 + 1
             }
-            (c1, u1)
-        }
-      log.info("received query_short_channel_ids with {} items, sent back {} channels and {} updates", shortChannelIds.array.size, channelCount, updatesCount)
+          }
+          if (flag_opt.map(QueryShortChannelIdsTlv.QueryFlagType.includeUpdate2).getOrElse(true)) {
+            d.updates.get(ChannelDesc(ca.shortChannelId, ca.nodeId2, ca.nodeId1)).foreach { u =>
+              transport ! u
+              numcu1 = numcu1 + 1
+            }
+          }
+          if (flag_opt.map(QueryShortChannelIdsTlv.QueryFlagType.includeNodeAnnouncement1).getOrElse(true) && !sent1.contains(ca.nodeId1)) {
+            d.nodes.get(ca.nodeId1).foreach { a =>
+              transport ! a
+              sent1 = sent1 + ca.nodeId1
+            }
+          }
+          if (flag_opt.map(QueryShortChannelIdsTlv.QueryFlagType.includeNodeAnnouncement2).getOrElse(true) && !sent1.contains(ca.nodeId2)) {
+            d.nodes.get(ca.nodeId2).foreach { a =>
+              transport ! a
+              sent1 = sent1 + ca.nodeId2
+            }
+          }
+          loop(tail, flags.drop(1), numca1, numcu1, sent1)
+      }
+
+      val (channelCount, updateCount, nodeCount) = loop(shortChannelIds.array, flags)
+      log.info("received query_short_channel_ids with {} items, sent back {} channels and {} updates and {} node announcements", shortChannelIds.array.size, channelCount, updateCount, nodeCount)
       transport ! ReplyShortChannelIdsEnd(chainHash, 1)
       stay
 
@@ -856,7 +889,8 @@ object Router {
   def computeFlag(channels: SortedMap[ShortChannelId, ChannelAnnouncement], updates: Map[ChannelDesc, ChannelUpdate])(
     shortChannelId: ShortChannelId,
     timestamps_opt: Option[ReplyChannelRangeTlv.Timestamps],
-    checksums_opt: Option[ReplyChannelRangeTlv.Checksums]): Long = {
+    checksums_opt: Option[ReplyChannelRangeTlv.Checksums],
+    includeNodeAnnouncements: Boolean): Long = {
     import QueryShortChannelIdsTlv.QueryFlagType
     var flag = 0L
     (timestamps_opt, checksums_opt) match {
@@ -884,8 +918,9 @@ object Router {
         // we know this channel: we only request their channel updates
         flag = QueryFlagType.INCLUDE_CHANNEL_UPDATE_1 | QueryFlagType.INCLUDE_CHANNEL_UPDATE_2
       case _ =>
-        // we don't know this channel: we request everything
-        flag = QueryFlagType.INCLUDE_CHANNEL_ANNOUNCEMENT | QueryFlagType.INCLUDE_CHANNEL_UPDATE_1 | QueryFlagType.INCLUDE_CHANNEL_UPDATE_2
+        // we don't know this channel: we request its channel announcement and updates
+        flag = QueryFlagType.INCLUDE_CHANNEL_ANNOUNCEMENT | QueryFlagType.INCLUDE_CHANNEL_UPDATE_1  | QueryFlagType.INCLUDE_CHANNEL_UPDATE_2
+        if (includeNodeAnnouncements) flag = flag | QueryFlagType.INCLUDE_NODE_ANNOUNCEMENT_1 | QueryFlagType.INCLUDE_NODE_ANNOUNCEMENT_2
     }
     flag
   }

eclair-core/src/main/scala/fr/acinq/eclair/wire/QueryShortChannelIdsTlv.scala

@@ -1,7 +1,7 @@
 package fr.acinq.eclair.wire
 
 import fr.acinq.eclair.UInt64
-import fr.acinq.eclair.wire.CommonCodecs.{shortchannelid, varint, varintoverflow}
+import fr.acinq.eclair.wire.CommonCodecs.{varint, varintoverflow}
 import scodec.Codec
 import scodec.codecs.{byte, discriminated, list, provide, variableSizeBytesLong, zlib}
 
@@ -20,13 +20,19 @@ object QueryShortChannelIdsTlv {
     val INCLUDE_CHANNEL_ANNOUNCEMENT: Long = 1
     val INCLUDE_CHANNEL_UPDATE_1: Long = 2
     val INCLUDE_CHANNEL_UPDATE_2: Long = 4
-    val INCLUDE_ALL: Long = (INCLUDE_CHANNEL_ANNOUNCEMENT | INCLUDE_CHANNEL_UPDATE_1 | INCLUDE_CHANNEL_UPDATE_2)
+    val INCLUDE_NODE_ANNOUNCEMENT_1: Long = 8
+    val INCLUDE_NODE_ANNOUNCEMENT_2: Long = 16
+    val INCLUDE_ALL: Long = (INCLUDE_CHANNEL_ANNOUNCEMENT | INCLUDE_CHANNEL_UPDATE_1 | INCLUDE_CHANNEL_UPDATE_2 | INCLUDE_NODE_ANNOUNCEMENT_1 | INCLUDE_NODE_ANNOUNCEMENT_2)
 
-    def includeAnnouncement(flag: Long) = (flag & INCLUDE_CHANNEL_ANNOUNCEMENT) != 0
+    def includeChannelAnnouncement(flag: Long) = (flag & INCLUDE_CHANNEL_ANNOUNCEMENT) != 0
 
     def includeUpdate1(flag: Long) = (flag & INCLUDE_CHANNEL_UPDATE_1) != 0
 
     def includeUpdate2(flag: Long) = (flag & INCLUDE_CHANNEL_UPDATE_2) != 0
+
+    def includeNodeAnnouncement1(flag: Long) = (flag & INCLUDE_NODE_ANNOUNCEMENT_1) != 0
+
+    def includeNodeAnnouncement2(flag: Long) = (flag & INCLUDE_NODE_ANNOUNCEMENT_2) != 0
   }
 
   val encodedQueryFlagsCodec: Codec[EncodedQueryFlags] =

eclair-core/src/test/scala/fr/acinq/eclair/TestConstants.scala

@@ -106,6 +106,7 @@ object TestConstants {
         randomizeRouteSelection = false,
         channelExcludeDuration = 60 seconds,
         routerBroadcastInterval = 5 seconds,
+        requestNodeAnnouncements = true,
         searchMaxFeeBase = Satoshi(21),
         searchMaxFeePct = 0.03,
         searchMaxCltv = 2016,
@@ -176,6 +177,7 @@ object TestConstants {
         randomizeRouteSelection = false,
         channelExcludeDuration = 60 seconds,
         routerBroadcastInterval = 5 seconds,
+        requestNodeAnnouncements = true,
         searchMaxFeeBase = Satoshi(21),
         searchMaxFeePct = 0.03,
         searchMaxCltv = 2016,

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

@@ -17,7 +17,6 @@
 package fr.acinq.eclair.router
 
 import fr.acinq.eclair.wire.ReplyChannelRangeTlv._
-import fr.acinq.eclair.wire._
 import fr.acinq.eclair.{MilliSatoshi, randomKey}
 import org.scalatest.FunSuite
 
@@ -27,21 +26,21 @@ import scala.compat.Platform
 
 class ChannelRangeQueriesSpec extends FunSuite {
 
-    test("compute flag tests") {
+  test("compute flag tests") {
 
     val now = Platform.currentTime / 1000
 
     val a = randomKey.publicKey
     val b = randomKey.publicKey
     val ab = RouteCalculationSpec.makeChannel(123466L, a, b)
-        val (ab1, uab1) = RouteCalculationSpec.makeUpdateShort(ab.shortChannelId, ab.nodeId1, ab.nodeId2, MilliSatoshi(0), 0, timestamp = now)
-        val (ab2, uab2) = RouteCalculationSpec.makeUpdateShort(ab.shortChannelId, ab.nodeId2, ab.nodeId1, MilliSatoshi(0), 0, timestamp = now)
+    val (ab1, uab1) = RouteCalculationSpec.makeUpdateShort(ab.shortChannelId, ab.nodeId1, ab.nodeId2, MilliSatoshi(0), 0, timestamp = now)
+    val (ab2, uab2) = RouteCalculationSpec.makeUpdateShort(ab.shortChannelId, ab.nodeId2, ab.nodeId1, MilliSatoshi(0), 0, timestamp = now)
 
     val c = randomKey.publicKey
     val d = randomKey.publicKey
     val cd = RouteCalculationSpec.makeChannel(451312L, c, d)
-        val (cd1, ucd1) = RouteCalculationSpec.makeUpdateShort(cd.shortChannelId, cd.nodeId1, cd.nodeId2, MilliSatoshi(0), 0, timestamp = now)
-        val (_, ucd2) = RouteCalculationSpec.makeUpdateShort(cd.shortChannelId, cd.nodeId2, cd.nodeId1, MilliSatoshi(0), 0, timestamp = now)
+    val (cd1, ucd1) = RouteCalculationSpec.makeUpdateShort(cd.shortChannelId, cd.nodeId1, cd.nodeId2, MilliSatoshi(0), 0, timestamp = now)
+    val (_, ucd2) = RouteCalculationSpec.makeUpdateShort(cd.shortChannelId, cd.nodeId2, cd.nodeId1, MilliSatoshi(0), 0, timestamp = now)
 
     val e = randomKey.publicKey
     val f = randomKey.publicKey
@@ -63,23 +62,23 @@ class ChannelRangeQueriesSpec extends FunSuite {
     assert(Router.getChannelDigestInfo(channels, updates)(ab.shortChannelId) == (Timestamps(now, now), Checksums(3297511804L, 3297511804L)))
 
     // no extended info but we know the channel: we ask for the updates
-    assert(Router.computeFlag(channels, updates)(ab.shortChannelId, None, None) === (INCLUDE_CHANNEL_UPDATE_1 | INCLUDE_CHANNEL_UPDATE_2).toByte)
+    assert(Router.computeFlag(channels, updates)(ab.shortChannelId, None, None, true) === (INCLUDE_CHANNEL_UPDATE_1 | INCLUDE_CHANNEL_UPDATE_2).toByte)
     // same checksums, newer timestamps: we don't ask anything
-    assert(Router.computeFlag(channels, updates)(ab.shortChannelId, Some(Timestamps(now + 1, now + 1)), Some(Checksums(3297511804L, 3297511804L))) === 0.toByte)
+    assert(Router.computeFlag(channels, updates)(ab.shortChannelId, Some(Timestamps(now + 1, now + 1)), Some(Checksums(3297511804L, 3297511804L)), true) === 0.toByte)
     // different checksums, newer timestamps: we ask for the updates
-    assert(Router.computeFlag(channels, updates)(ab.shortChannelId, Some(Timestamps(now + 1, now)), Some(Checksums(154654604, 3297511804L))) === INCLUDE_CHANNEL_UPDATE_1)
-    assert(Router.computeFlag(channels, updates)(ab.shortChannelId, Some(Timestamps(now, now + 1)), Some(Checksums(3297511804L, 45664546))) === INCLUDE_CHANNEL_UPDATE_2)
-    assert(Router.computeFlag(channels, updates)(ab.shortChannelId, Some(Timestamps(now + 1, now + 1)), Some(Checksums(154654604, 45664546+6))) === (INCLUDE_CHANNEL_UPDATE_1 | INCLUDE_CHANNEL_UPDATE_2).toByte)
+    assert(Router.computeFlag(channels, updates)(ab.shortChannelId, Some(Timestamps(now + 1, now)), Some(Checksums(154654604, 3297511804L)), true) === INCLUDE_CHANNEL_UPDATE_1)
+    assert(Router.computeFlag(channels, updates)(ab.shortChannelId, Some(Timestamps(now, now + 1)), Some(Checksums(3297511804L, 45664546)), true) === INCLUDE_CHANNEL_UPDATE_2)
+    assert(Router.computeFlag(channels, updates)(ab.shortChannelId, Some(Timestamps(now + 1, now + 1)), Some(Checksums(154654604, 45664546 + 6)), true) === (INCLUDE_CHANNEL_UPDATE_1 | INCLUDE_CHANNEL_UPDATE_2).toByte)
     // different checksums, older timestamps: we don't ask anything
-    assert(Router.computeFlag(channels, updates)(ab.shortChannelId, Some(Timestamps(now - 1, now)), Some(Checksums(154654604, 3297511804L))) === 0.toByte)
-    assert(Router.computeFlag(channels, updates)(ab.shortChannelId, Some(Timestamps(now, now - 1)), Some(Checksums(3297511804L, 45664546))) === 0.toByte)
-    assert(Router.computeFlag(channels, updates)(ab.shortChannelId, Some(Timestamps(now - 1, now - 1)), Some(Checksums(154654604, 45664546))) === 0.toByte)
+    assert(Router.computeFlag(channels, updates)(ab.shortChannelId, Some(Timestamps(now - 1, now)), Some(Checksums(154654604, 3297511804L)), true) === 0.toByte)
+    assert(Router.computeFlag(channels, updates)(ab.shortChannelId, Some(Timestamps(now, now - 1)), Some(Checksums(3297511804L, 45664546)), true) === 0.toByte)
+    assert(Router.computeFlag(channels, updates)(ab.shortChannelId, Some(Timestamps(now - 1, now - 1)), Some(Checksums(154654604, 45664546)), true) === 0.toByte)
 
     // missing channel update: we ask for it
-    assert(Router.computeFlag(channels, updates)(cd.shortChannelId, Some(Timestamps(now, now)), Some(Checksums(3297511804L, 3297511804L))) === INCLUDE_CHANNEL_UPDATE_2)
+    assert(Router.computeFlag(channels, updates)(cd.shortChannelId, Some(Timestamps(now, now)), Some(Checksums(3297511804L, 3297511804L)), true) === INCLUDE_CHANNEL_UPDATE_2)
 
     // unknown channel: we ask everything
-    assert(Router.computeFlag(channels, updates)(ef.shortChannelId, None, None) === INCLUDE_ALL)
-
+    assert(Router.computeFlag(channels, updates)(ef.shortChannelId, None, None, false) === (INCLUDE_CHANNEL_ANNOUNCEMENT | INCLUDE_CHANNEL_UPDATE_1 | INCLUDE_CHANNEL_UPDATE_2))
+    assert(Router.computeFlag(channels, updates)(ef.shortChannelId, None, None, true) === (INCLUDE_CHANNEL_ANNOUNCEMENT | INCLUDE_CHANNEL_UPDATE_1 | INCLUDE_CHANNEL_UPDATE_2 | INCLUDE_NODE_ANNOUNCEMENT_1 | INCLUDE_NODE_ANNOUNCEMENT_2))
   }
 }