Add support for option_compression

Implement lightning/bolts#825

This lets us specify which compression algorithms we support and use one
that our peer supports as well when syncing the network graph.

docs/release-notes/eclair-vnext.md

@@ -24,6 +24,14 @@ Eclair now supports the feature `option_onion_messages`. If this feature is enab
 It can also send onion messages with the `sendonionmessage` API.
 Messages sent to Eclair will be ignored.
 
+### Support for `option_compression`
+
+Eclair now supports the `option_compression` feature as specified in https://github.com/lightning/bolts/pull/825.
+Eclair will announce what compression algorithms it supports for routing sync, and will only use compression algorithms supported by its peers when forwarding gossip.
+
+If you were overriding the default `eclair.router.sync.encoding-type` in your `eclair.conf`, you need to update your configuration.
+This field has been renamed `eclair.router.sync.preferred-compression-algorithm` and defaults to `zlib`.
+
 ### API changes
 
 #### Timestamps

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

@@ -55,6 +55,7 @@ eclair {
     // Do not enable option_anchor_outputs unless you really know what you're doing.
     option_anchor_outputs = disabled
     option_anchors_zero_fee_htlc_tx = disabled
+    option_compression = optional
     option_shutdown_anysegwit = optional
     option_onion_messages = disabled
     trampoline_payment = disabled
@@ -218,7 +219,7 @@ eclair {
 
     sync {
       request-node-announcements = true // if true we will ask for node announcements when we receive channel ids that we don't know
-      encoding-type = zlib // encoding for short_channel_ids and timestamps in query channel sync messages; other possible value is "uncompressed"
+      preferred-compression-algorithm = zlib // encoding for short_channel_ids and timestamps in query channel sync messages; other possible value is "uncompressed"
       channel-range-chunk-size = 1500 // max number of short_channel_ids (+ timestamps + checksums) in reply_channel_range *do not change this unless you know what you are doing*
       channel-query-chunk-size = 100 // max number of short_channel_ids in query_short_channel_ids *do not change this unless you know what you are doing*
     }

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

@@ -198,6 +198,11 @@ object Features {
     val mandatory = 22
   }
 
+  case object CompressionSupport extends Feature {
+    val rfcName = "option_compression"
+    val mandatory = 24
+  }
+
   case object ShutdownAnySegwit extends Feature {
     val rfcName = "option_shutdown_anysegwit"
     val mandatory = 26
@@ -235,6 +240,7 @@ object Features {
     StaticRemoteKey,
     AnchorOutputs,
     AnchorOutputsZeroFeeHtlcTx,
+    CompressionSupport,
     ShutdownAnySegwit,
     OnionMessages,
     KeySend

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

@@ -32,7 +32,7 @@ import fr.acinq.eclair.router.Graph.{HeuristicsConstants, WeightRatios}
 import fr.acinq.eclair.router.PathFindingExperimentConf
 import fr.acinq.eclair.router.Router.{MultiPartParams, PathFindingConf, RouterConf, SearchBoundaries}
 import fr.acinq.eclair.tor.Socks5ProxyParams
-import fr.acinq.eclair.wire.protocol.{Color, EncodingType, NodeAddress}
+import fr.acinq.eclair.wire.protocol.{Color, CompressionAlgorithm, NodeAddress}
 import grizzled.slf4j.Logging
 import scodec.bits.ByteVector
 
@@ -217,6 +217,7 @@ object NodeParams extends Logging {
       "router.path-finding.ratio-channel-capacity" -> "router.path-finding.default.ratios.channel-capacity",
       "router.path-finding.hop-cost-base-msat" -> "router.path-finding.default.hop-cost.fee-base-msat",
       "router.path-finding.hop-cost-millionths" -> "router.path-finding.default.hop-cost.fee-proportional-millionths",
+      "router.sync.encoding-type" -> "router.sync.preferred-compression-algorithm"
     )
     deprecatedKeyPaths.foreach {
       case (old, new_) => require(!config.hasPath(old), s"configuration key '$old' has been replaced by '$new_'")
@@ -353,7 +354,6 @@ object NodeParams extends Logging {
       experimentName = name,
       experimentPercentage = config.getInt("percentage"))
 
-
     def getPathFindingExperimentConf(config: Config): PathFindingExperimentConf = {
       val experiments = config.root.asScala.keys.map(name => name -> getPathFindingConf(config.getConfig(name), name))
       PathFindingExperimentConf(experiments.toMap)
@@ -364,9 +364,9 @@ object NodeParams extends Logging {
       case "stop" => UnhandledExceptionStrategy.Stop
     }
 
-    val routerSyncEncodingType = config.getString("router.sync.encoding-type") match {
-      case "uncompressed" => EncodingType.UNCOMPRESSED
-      case "zlib" => EncodingType.COMPRESSED_ZLIB
+    val routerSyncPreferredCompression = config.getString("router.sync.preferred-compression-algorithm") match {
+      case "uncompressed" => CompressionAlgorithm.Uncompressed
+      case "zlib" => CompressionAlgorithm.ZlibDeflate
     }
 
     NodeParams(
@@ -456,7 +456,7 @@ object NodeParams extends Logging {
         routerBroadcastInterval = FiniteDuration(config.getDuration("router.broadcast-interval").getSeconds, TimeUnit.SECONDS),
         networkStatsRefreshInterval = FiniteDuration(config.getDuration("router.network-stats-interval").getSeconds, TimeUnit.SECONDS),
         requestNodeAnnouncements = config.getBoolean("router.sync.request-node-announcements"),
-        encodingType = routerSyncEncodingType,
+        preferredCompression = routerSyncPreferredCompression,
         channelRangeChunkSize = config.getInt("router.sync.channel-range-chunk-size"),
         channelQueryChunkSize = config.getInt("router.sync.channel-query-chunk-size"),
         pathFindingExperimentConf = getPathFindingExperimentConf(config.getConfig("router.path-finding.experiments"))

eclair-core/src/main/scala/fr/acinq/eclair/io/Peer.scala

@@ -452,7 +452,7 @@ object Peer {
   case object GetPeerInfo
   case class PeerInfo(nodeId: PublicKey, state: String, address: Option[InetSocketAddress], channels: Int)
 
-  case class PeerRoutingMessage(peerConnection: ActorRef, remoteNodeId: PublicKey, message: RoutingMessage) extends RemoteTypes
+  case class PeerRoutingMessage(peerConnection: ActorRef, remoteNodeId: PublicKey, remoteInit: protocol.Init, message: RoutingMessage) extends RemoteTypes
 
   /**
    * Dedicated command for outgoing messages for logging purposes.

eclair-core/src/main/scala/fr/acinq/eclair/io/PeerConnection.scala

@@ -101,7 +101,7 @@ class PeerConnection(keyPair: KeyPair, conf: PeerConnection.Conf, switchboard: A
       d.transport ! TransportHandler.Listener(self)
       Metrics.PeerConnectionsConnecting.withTag(Tags.ConnectionState, Tags.ConnectionStates.Initializing).increment()
       log.info(s"using features=$localFeatures")
-      val localInit = protocol.Init(localFeatures, TlvStream(InitTlv.Networks(chainHash :: Nil)))
+      val localInit = protocol.Init(localFeatures, TlvStream(InitTlv.Networks(chainHash :: Nil), InitTlv.CompressionAlgorithms(Set(CompressionAlgorithm.Uncompressed, CompressionAlgorithm.ZlibDeflate))))
       d.transport ! localInit
       startSingleTimer(INIT_TIMER, InitTimeout, conf.initTimeout)
       goto(INITIALIZING) using InitializingData(chainHash, d.pendingAuth, d.remoteNodeId, d.transport, peer, localInit, doSync)
@@ -234,7 +234,7 @@ class PeerConnection(keyPair: KeyPair, conf: PeerConnection.Conf, switchboard: A
 
       case Event(DelayedRebroadcast(rebroadcast), d: ConnectedData) =>
 
-        val thisRemote = RemoteGossip(self, d.remoteNodeId)
+        val thisRemote = RemoteGossip(self, d.remoteNodeId, d.remoteInit)
 
         /**
          * Send and count in a single iteration
@@ -285,7 +285,7 @@ class PeerConnection(keyPair: KeyPair, conf: PeerConnection.Conf, switchboard: A
             d.transport ! TransportHandler.ReadAck(msg)
           case _ =>
             // Note: we don't ack messages here because we don't want them to be stacked in the router's mailbox
-            router ! Peer.PeerRoutingMessage(self, d.remoteNodeId, msg)
+            router ! Peer.PeerRoutingMessage(self, d.remoteNodeId, d.remoteInit, msg)
         }
         stay()
 
@@ -347,7 +347,8 @@ class PeerConnection(keyPair: KeyPair, conf: PeerConnection.Conf, switchboard: A
       case Event(DoSync(replacePrevious), d: ConnectedData) =>
         val canUseChannelRangeQueries = Features.canUseFeature(d.localInit.features, d.remoteInit.features, Features.ChannelRangeQueries)
         val canUseChannelRangeQueriesEx = Features.canUseFeature(d.localInit.features, d.remoteInit.features, Features.ChannelRangeQueriesExtended)
-        if (canUseChannelRangeQueries || canUseChannelRangeQueriesEx) {
+        val hasCompatibleCompression = CompressionAlgorithm.select(d.localInit.compressionAlgorithms, d.remoteInit.compressionAlgorithms).nonEmpty
+        if ((canUseChannelRangeQueries || canUseChannelRangeQueriesEx) && hasCompatibleCompression) {
           val flags_opt = if (canUseChannelRangeQueriesEx) Some(QueryChannelRangeTlv.QueryFlags(QueryChannelRangeTlv.QueryFlags.WANT_ALL)) else None
           log.info(s"sending sync channel range query with flags_opt=$flags_opt replacePrevious=$replacePrevious")
           router ! SendChannelQuery(d.chainHash, d.remoteNodeId, self, replacePrevious, flags_opt)

eclair-core/src/main/scala/fr/acinq/eclair/remote/EclairInternalsSerializer.scala

@@ -83,17 +83,19 @@ object EclairInternalsSerializer {
       ("experimentPercentage" | int32)).as[PathFindingConf]
 
   val pathFindingExperimentConfCodec: Codec[PathFindingExperimentConf] = (
-    ("experiments" | listOfN(int32, pathFindingConfCodec).xmap[Map[String, PathFindingConf]](_.map(e => (e.experimentName -> e)).toMap, _.values.toList))
+    "experiments" | listOfN(int32, pathFindingConfCodec).xmap[Map[String, PathFindingConf]](_.map(e => e.experimentName -> e).toMap, _.values.toList)
     ).as[PathFindingExperimentConf]
 
+  private val compressionAlgorithmCodec: Codec[CompressionAlgorithm] = discriminated[CompressionAlgorithm].by(uint8)
+    .typecase(CompressionAlgorithm.Uncompressed.bitPosition, provide(CompressionAlgorithm.Uncompressed))
+    .typecase(CompressionAlgorithm.ZlibDeflate.bitPosition, provide(CompressionAlgorithm.ZlibDeflate))
+
   val routerConfCodec: Codec[RouterConf] = (
     ("channelExcludeDuration" | finiteDurationCodec) ::
       ("routerBroadcastInterval" | finiteDurationCodec) ::
       ("networkStatsRefreshInterval" | finiteDurationCodec) ::
       ("requestNodeAnnouncements" | bool(8)) ::
-      ("encodingType" | discriminated[EncodingType].by(uint8)
-        .typecase(0, provide(EncodingType.UNCOMPRESSED))
-        .typecase(1, provide(EncodingType.COMPRESSED_ZLIB))) ::
+      ("preferredCompression" | compressionAlgorithmCodec) ::
       ("channelRangeChunkSize" | int32) ::
       ("channelQueryChunkSize" | int32) ::
       ("pathFindingExperimentConf" | pathFindingExperimentConfCodec)).as[RouterConf]
@@ -166,6 +168,7 @@ object EclairInternalsSerializer {
   def peerRoutingMessageCodec(system: ExtendedActorSystem): Codec[PeerRoutingMessage] = (
     ("peerConnection" | actorRefCodec(system)) ::
       ("remoteNodeId" | publicKey) ::
+      ("remoteInit" | lengthPrefixedInitCodec) ::
       ("msg" | lengthPrefixedLightningMessageCodec.downcast[RoutingMessage])).as[PeerRoutingMessage]
 
   val singleChannelDiscoveredCodec: Codec[SingleChannelDiscovered] = (lengthPrefixedChannelAnnouncementCodec :: satoshi :: optional(bool(8), lengthPrefixedChannelUpdateCodec) :: optional(bool(8), lengthPrefixedChannelUpdateCodec)).as[SingleChannelDiscovered]

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

@@ -223,13 +223,13 @@ class Router(val nodeParams: NodeParams, watcher: typed.ActorRef[ZmqWatcher.Comm
       stay() using RouteCalculation.handleRouteRequest(d, nodeParams.routerConf, nodeParams.currentBlockHeight, r)
 
     // Warning: order matters here, this must be the first match for HasChainHash messages !
-    case Event(PeerRoutingMessage(_, _, routingMessage: HasChainHash), _) if routingMessage.chainHash != nodeParams.chainHash =>
+    case Event(PeerRoutingMessage(_, _, _, routingMessage: HasChainHash), _) if routingMessage.chainHash != nodeParams.chainHash =>
       sender() ! TransportHandler.ReadAck(routingMessage)
       log.warning("message {} for wrong chain {}, we're on {}", routingMessage, routingMessage.chainHash, nodeParams.chainHash)
       stay()
 
-    case Event(PeerRoutingMessage(peerConnection, remoteNodeId, c: ChannelAnnouncement), d) =>
-      stay() using Validation.handleChannelAnnouncement(d, nodeParams.db.network, watcher, RemoteGossip(peerConnection, remoteNodeId), c)
+    case Event(PeerRoutingMessage(peerConnection, remoteNodeId, remoteInit, c: ChannelAnnouncement), d) =>
+      stay() using Validation.handleChannelAnnouncement(d, nodeParams.db.network, watcher, RemoteGossip(peerConnection, remoteNodeId, remoteInit), c)
 
     case Event(r: ValidateResult, d) =>
       stay() using Validation.handleChannelValidationResponse(d, nodeParams, watcher, r)
@@ -240,14 +240,14 @@ class Router(val nodeParams: NodeParams, watcher: typed.ActorRef[ZmqWatcher.Comm
     case Event(n: NodeAnnouncement, d: Data) =>
       stay() using Validation.handleNodeAnnouncement(d, nodeParams.db.network, Set(LocalGossip), n)
 
-    case Event(PeerRoutingMessage(peerConnection, remoteNodeId, n: NodeAnnouncement), d: Data) =>
-      stay() using Validation.handleNodeAnnouncement(d, nodeParams.db.network, Set(RemoteGossip(peerConnection, remoteNodeId)), n)
+    case Event(PeerRoutingMessage(peerConnection, remoteNodeId, remoteInit, n: NodeAnnouncement), d: Data) =>
+      stay() using Validation.handleNodeAnnouncement(d, nodeParams.db.network, Set(RemoteGossip(peerConnection, remoteNodeId, remoteInit)), n)
 
     case Event(u: ChannelUpdate, d: Data) =>
       stay() using Validation.handleChannelUpdate(d, nodeParams.db.network, nodeParams.routerConf, Right(RemoteChannelUpdate(u, Set(LocalGossip))))
 
-    case Event(PeerRoutingMessage(peerConnection, remoteNodeId, u: ChannelUpdate), d) =>
-      stay() using Validation.handleChannelUpdate(d, nodeParams.db.network, nodeParams.routerConf, Right(RemoteChannelUpdate(u, Set(RemoteGossip(peerConnection, remoteNodeId)))))
+    case Event(PeerRoutingMessage(peerConnection, remoteNodeId, remoteInit, u: ChannelUpdate), d) =>
+      stay() using Validation.handleChannelUpdate(d, nodeParams.db.network, nodeParams.routerConf, Right(RemoteChannelUpdate(u, Set(RemoteGossip(peerConnection, remoteNodeId, remoteInit)))))
 
     case Event(lcu: LocalChannelUpdate, d: Data) =>
       stay() using Validation.handleLocalChannelUpdate(d, nodeParams.db.network, nodeParams.routerConf, nodeParams.nodeId, watcher, lcu)
@@ -261,19 +261,19 @@ class Router(val nodeParams: NodeParams, watcher: typed.ActorRef[ZmqWatcher.Comm
     case Event(s: SendChannelQuery, d) =>
       stay() using Sync.handleSendChannelQuery(d, s)
 
-    case Event(PeerRoutingMessage(peerConnection, remoteNodeId, q: QueryChannelRange), d) =>
-      Sync.handleQueryChannelRange(d.channels, nodeParams.routerConf, RemoteGossip(peerConnection, remoteNodeId), q)
+    case Event(PeerRoutingMessage(peerConnection, remoteNodeId, remoteInit, q: QueryChannelRange), d) =>
+      Sync.handleQueryChannelRange(d.channels, nodeParams.routerConf, RemoteGossip(peerConnection, remoteNodeId, remoteInit), 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, remoteInit, r: ReplyChannelRange), d) =>
+      stay() using Sync.handleReplyChannelRange(d, nodeParams.routerConf, RemoteGossip(peerConnection, remoteNodeId, remoteInit), r)
 
-    case Event(PeerRoutingMessage(peerConnection, remoteNodeId, q: QueryShortChannelIds), d) =>
-      Sync.handleQueryShortChannelIds(d.nodes, d.channels, RemoteGossip(peerConnection, remoteNodeId), q)
+    case Event(PeerRoutingMessage(peerConnection, remoteNodeId, remoteInit, q: QueryShortChannelIds), d) =>
+      Sync.handleQueryShortChannelIds(d.nodes, d.channels, RemoteGossip(peerConnection, remoteNodeId, remoteInit), q)
       stay()
 
-    case Event(PeerRoutingMessage(peerConnection, remoteNodeId, r: ReplyShortChannelIdsEnd), d) =>
-      stay() using Sync.handleReplyShortChannelIdsEnd(d, RemoteGossip(peerConnection, remoteNodeId), r)
+    case Event(PeerRoutingMessage(peerConnection, remoteNodeId, remoteInit, r: ReplyShortChannelIdsEnd), d) =>
+      stay() using Sync.handleReplyShortChannelIdsEnd(d, RemoteGossip(peerConnection, remoteNodeId, remoteInit), r)
 
   }
 
@@ -322,7 +322,7 @@ object Router {
                         routerBroadcastInterval: FiniteDuration,
                         networkStatsRefreshInterval: FiniteDuration,
                         requestNodeAnnouncements: Boolean,
-                        encodingType: EncodingType,
+                        preferredCompression: CompressionAlgorithm,
                         channelRangeChunkSize: Int,
                         channelQueryChunkSize: Int,
                         pathFindingExperimentConf: PathFindingExperimentConf)
@@ -552,7 +552,7 @@ object Router {
   // @formatter:off
   sealed trait GossipOrigin
   /** Gossip that we received from a remote peer. */
-  case class RemoteGossip(peerConnection: ActorRef, nodeId: PublicKey) extends GossipOrigin
+  case class RemoteGossip(peerConnection: ActorRef, nodeId: PublicKey, remoteInit: Init) extends GossipOrigin
   /** Gossip that was generated by our node. */
   case object LocalGossip extends GossipOrigin