Rework router data structures

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

@@ -39,7 +39,7 @@ object DBCompatChecker extends Logging {
     * @param nodeParams
     */
   def checkNetworkDBCompatibility(nodeParams: NodeParams): Unit =
-    Try(nodeParams.db.network.listChannels(), nodeParams.db.network.listNodes(), nodeParams.db.network.listChannelUpdates()) match {
+    Try(nodeParams.db.network.listChannels(), nodeParams.db.network.listNodes()) match {
       case Success(_) => {}
       case Failure(_) => throw IncompatibleNetworkDBException
     }

eclair-core/src/main/scala/fr/acinq/eclair/db/NetworkDb.scala

@@ -19,8 +19,11 @@ package fr.acinq.eclair.db
 import fr.acinq.bitcoin.Crypto.PublicKey
 import fr.acinq.bitcoin.{ByteVector32, Satoshi}
 import fr.acinq.eclair.ShortChannelId
+import fr.acinq.eclair.router.PublicChannel
 import fr.acinq.eclair.wire.{ChannelAnnouncement, ChannelUpdate, NodeAnnouncement}
 
+import scala.collection.immutable.SortedMap
+
 trait NetworkDb {
 
   def addNode(n: NodeAnnouncement)
@@ -35,22 +38,13 @@ trait NetworkDb {
 
   def addChannel(c: ChannelAnnouncement, txid: ByteVector32, capacity: Satoshi)
 
-  def removeChannel(shortChannelId: ShortChannelId) = removeChannels(Seq(shortChannelId))
-
-  /**
-    * This method removes channel announcements and associated channel updates for a list of channel ids
-    *
-    * @param shortChannelIds list of short channel ids
-    */
-  def removeChannels(shortChannelIds: Iterable[ShortChannelId])
+  def updateChannel(u: ChannelUpdate)
 
-  def listChannels(): Map[ChannelAnnouncement, (ByteVector32, Satoshi)]
+  def removeChannel(shortChannelId: ShortChannelId) = removeChannels(Set(shortChannelId))
 
-  def addChannelUpdate(u: ChannelUpdate)
-
-  def updateChannelUpdate(u: ChannelUpdate)
+  def removeChannels(shortChannelIds: Iterable[ShortChannelId])
 
-  def listChannelUpdates(): Seq[ChannelUpdate]
+  def listChannels(): SortedMap[ShortChannelId, PublicChannel]
 
   def addToPruned(shortChannelIds: Iterable[ShortChannelId]): Unit
 

eclair-core/src/main/scala/fr/acinq/eclair/db/sqlite/SqliteNetworkDb.scala

@@ -21,25 +21,39 @@ import java.sql.Connection
 import fr.acinq.bitcoin.{ByteVector32, Crypto, Satoshi}
 import fr.acinq.eclair.ShortChannelId
 import fr.acinq.eclair.db.NetworkDb
-import fr.acinq.eclair.router.Announcements
+import fr.acinq.eclair.router.PublicChannel
 import fr.acinq.eclair.wire.LightningMessageCodecs.{channelAnnouncementCodec, channelUpdateCodec, nodeAnnouncementCodec}
 import fr.acinq.eclair.wire.{ChannelAnnouncement, ChannelUpdate, NodeAnnouncement}
-import scodec.bits.BitVector
+import grizzled.slf4j.Logging
 
-class SqliteNetworkDb(sqlite: Connection) extends NetworkDb {
+import scala.collection.immutable.SortedMap
+
+class SqliteNetworkDb(sqlite: Connection) extends NetworkDb with Logging {
 
   import SqliteUtils._
+  import SqliteUtils.ExtendedResultSet._
 
   val DB_NAME = "network"
-  val CURRENT_VERSION = 1
+  val CURRENT_VERSION = 2
 
   using(sqlite.createStatement()) { statement =>
-    require(getVersion(statement, DB_NAME, CURRENT_VERSION) == CURRENT_VERSION, s"incompatible version of $DB_NAME DB found") // there is only one version currently deployed
-    statement.execute("PRAGMA foreign_keys = ON")
+    getVersion(statement, DB_NAME, CURRENT_VERSION) match {
+      case 1 =>
+        // channel_update are cheap to retrieve, so let's just wipe them out and they'll get resynced
+        statement.execute("PRAGMA foreign_keys = ON")
+        logger.warn("migrating network db version 1->2")
+        statement.executeUpdate("ALTER TABLE channels RENAME COLUMN data TO channel_announcement")
+        statement.executeUpdate("ALTER TABLE channels ADD COLUMN channel_update_1 BLOB NULL")
+        statement.executeUpdate("ALTER TABLE channels ADD COLUMN channel_update_2 BLOB NULL")
+        statement.executeUpdate("DROP TABLE channel_updates")
+        statement.execute("PRAGMA foreign_keys = OFF")
+        setVersion(statement, DB_NAME, CURRENT_VERSION)
+        logger.warn("migration complete")
+      case 2 => () // nothing to do
+      case unknown => throw new IllegalArgumentException(s"unknown version $unknown for network db")
+    }
     statement.executeUpdate("CREATE TABLE IF NOT EXISTS nodes (node_id BLOB NOT NULL PRIMARY KEY, data BLOB NOT NULL)")
-    statement.executeUpdate("CREATE TABLE IF NOT EXISTS channels (short_channel_id INTEGER NOT NULL PRIMARY KEY, txid STRING NOT NULL, data BLOB NOT NULL, capacity_sat INTEGER NOT NULL)")
-    statement.executeUpdate("CREATE TABLE IF NOT EXISTS channel_updates (short_channel_id INTEGER NOT NULL, node_flag INTEGER NOT NULL, data BLOB NOT NULL, PRIMARY KEY(short_channel_id, node_flag), FOREIGN KEY(short_channel_id) REFERENCES channels(short_channel_id))")
-    statement.executeUpdate("CREATE INDEX IF NOT EXISTS channel_updates_idx ON channel_updates(short_channel_id)")
+    statement.executeUpdate("CREATE TABLE IF NOT EXISTS channels (short_channel_id INTEGER NOT NULL PRIMARY KEY, txid STRING NOT NULL, channel_announcement BLOB NOT NULL, capacity_sat INTEGER NOT NULL, channel_update_1 BLOB NULL, channel_update_2 BLOB NULL)")
     statement.executeUpdate("CREATE TABLE IF NOT EXISTS pruned (short_channel_id INTEGER NOT NULL PRIMARY KEY)")
   }
 
@@ -82,7 +96,7 @@ class SqliteNetworkDb(sqlite: Connection) extends NetworkDb {
   }
 
   override def addChannel(c: ChannelAnnouncement, txid: ByteVector32, capacity: Satoshi): Unit = {
-    using(sqlite.prepareStatement("INSERT OR IGNORE INTO channels VALUES (?, ?, ?, ?)")) { statement =>
+    using(sqlite.prepareStatement("INSERT OR IGNORE INTO channels VALUES (?, ?, ?, ?, NULL, NULL)")) { statement =>
       statement.setLong(1, c.shortChannelId.toLong)
       statement.setString(2, txid.toHex)
       statement.setBytes(3, channelAnnouncementCodec.encode(c).require.toByteArray)
@@ -91,56 +105,39 @@ class SqliteNetworkDb(sqlite: Connection) extends NetworkDb {
     }
   }
 
-  override def removeChannels(shortChannelIds: Iterable[ShortChannelId]): Unit = {
-
-    def removeChannelsInternal(shortChannelIds: Iterable[ShortChannelId]): Unit = {
-      val ids = shortChannelIds.map(_.toLong).mkString(",")
-      using(sqlite.createStatement) { statement =>
-        statement.execute("BEGIN TRANSACTION")
-        statement.executeUpdate(s"DELETE FROM channel_updates WHERE short_channel_id IN ($ids)")
-        statement.executeUpdate(s"DELETE FROM channels WHERE short_channel_id IN ($ids)")
-        statement.execute("COMMIT TRANSACTION")
-      }
+  override def updateChannel(u: ChannelUpdate): Unit = {
+    val column = if (u.isNode1) "channel_update_1" else "channel_update_2"
+    using(sqlite.prepareStatement(s"UPDATE channels SET $column=? WHERE short_channel_id=?")) { statement =>
+      statement.setBytes(1, channelUpdateCodec.encode(u).require.toByteArray)
+      statement.setLong(2, u.shortChannelId.toLong)
+      statement.executeUpdate()
     }
-
-    // remove channels by batch of 1000
-    shortChannelIds.grouped(1000).foreach(removeChannelsInternal)
   }
 
-  override def listChannels(): Map[ChannelAnnouncement, (ByteVector32, Satoshi)] = {
+  override def listChannels(): SortedMap[ShortChannelId, PublicChannel] = {
     using(sqlite.createStatement()) { statement =>
-      val rs = statement.executeQuery("SELECT data, txid, capacity_sat FROM channels")
-      var m: Map[ChannelAnnouncement, (ByteVector32, Satoshi)] = Map()
+      val rs = statement.executeQuery("SELECT channel_announcement, txid, capacity_sat, channel_update_1, channel_update_2 FROM channels")
+      var m = SortedMap.empty[ShortChannelId, PublicChannel]
       while (rs.next()) {
-        m += (channelAnnouncementCodec.decode(BitVector(rs.getBytes("data"))).require.value ->
-          (ByteVector32.fromValidHex(rs.getString("txid")), Satoshi(rs.getLong("capacity_sat"))))
+        val ann = channelAnnouncementCodec.decode(rs.getBitVectorOpt("channel_announcement").get).require.value
+        val txId = ByteVector32.fromValidHex(rs.getString("txid"))
+        val capacity = rs.getLong("capacity_sat")
+        val channel_update_1_opt = rs.getBitVectorOpt("channel_update_1").map(channelUpdateCodec.decode(_).require.value)
+        val channel_update_2_opt = rs.getBitVectorOpt("channel_update_2").map(channelUpdateCodec.decode(_).require.value)
+        m = m + (ann.shortChannelId -> PublicChannel(ann, txId, Satoshi(capacity), channel_update_1_opt, channel_update_2_opt))
       }
       m
     }
   }
 
-  override def addChannelUpdate(u: ChannelUpdate): Unit = {
-    using(sqlite.prepareStatement("INSERT OR IGNORE INTO channel_updates VALUES (?, ?, ?)")) { statement =>
-      statement.setLong(1, u.shortChannelId.toLong)
-      statement.setBoolean(2, Announcements.isNode1(u.channelFlags))
-      statement.setBytes(3, channelUpdateCodec.encode(u).require.toByteArray)
-      statement.executeUpdate()
-    }
-  }
-
-  override def updateChannelUpdate(u: ChannelUpdate): Unit = {
-    using(sqlite.prepareStatement("UPDATE channel_updates SET data=? WHERE short_channel_id=? AND node_flag=?")) { statement =>
-      statement.setBytes(1, channelUpdateCodec.encode(u).require.toByteArray)
-      statement.setLong(2, u.shortChannelId.toLong)
-      statement.setBoolean(3, Announcements.isNode1(u.channelFlags))
-      statement.executeUpdate()
-    }
-  }
-
-  override def listChannelUpdates(): Seq[ChannelUpdate] = {
-    using(sqlite.createStatement()) { statement =>
-      val rs = statement.executeQuery("SELECT data FROM channel_updates")
-      codecSequence(rs, channelUpdateCodec)
+  override def removeChannels(shortChannelIds: Iterable[ShortChannelId]): Unit = {
+    using(sqlite.createStatement) { statement =>
+    shortChannelIds
+      .grouped(1000) // remove channels by batch of 1000
+      .foreach {group =>
+        val ids = shortChannelIds.map(_.toLong).mkString(",")
+        statement.executeUpdate(s"DELETE FROM channels WHERE short_channel_id IN ($ids)")
+      }
     }
   }
 

eclair-core/src/main/scala/fr/acinq/eclair/db/sqlite/SqliteUtils.scala

@@ -123,6 +123,8 @@ object SqliteUtils {
 
   case class ExtendedResultSet(rs: ResultSet) {
 
+    def getBitVectorOpt(columnLabel: String): Option[BitVector] = Option(rs.getBytes(columnLabel)).map(BitVector(_))
+
     def getByteVector(columnLabel: String): ByteVector = ByteVector(rs.getBytes(columnLabel))
 
     def getByteVector32(columnLabel: String): ByteVector32 = ByteVector32(ByteVector(rs.getBytes(columnLabel)))

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

@@ -334,7 +334,7 @@ class Peer(val nodeParams: NodeParams, remoteNodeId: PublicKey, authenticator: A
       // we won't clean it up, but we won't remember the temporary id on channel termination
       stay using d.copy(channels = d.channels + (FinalChannelId(channelId) -> channel))
 
-    case Event(RoutingState(channels, updates, nodes), d: ConnectedData) =>
+    case Event(RoutingState(channels, nodes), d: ConnectedData) =>
       // let's send the messages
       def send(announcements: Iterable[_ <: LightningMessage]) = announcements.foldLeft(0) {
         case (c, ann) =>
@@ -343,9 +343,9 @@ class Peer(val nodeParams: NodeParams, remoteNodeId: PublicKey, authenticator: A
       }
 
       log.info(s"sending all announcements to {}", remoteNodeId)
-      val channelsSent = send(channels)
+      val channelsSent = send(channels.map(_.ann))
       val nodesSent = send(nodes)
-      val updatesSent = send(updates)
+      val updatesSent = send(channels.flatMap(c => c.update_1_opt.toSeq ++ c.update_2_opt.toSeq))
       log.info(s"sent all announcements to {}: channels={} updates={} nodes={}", remoteNodeId, channelsSent, updatesSent, nodesSent)
       stay
 

eclair-core/src/main/scala/fr/acinq/eclair/payment/Autoprobe.scala

@@ -20,7 +20,7 @@ import akka.actor.{Actor, ActorLogging, ActorRef, Props}
 import fr.acinq.bitcoin.Crypto.PublicKey
 import fr.acinq.eclair.crypto.Sphinx.DecryptedFailurePacket
 import fr.acinq.eclair.payment.PaymentLifecycle.{PaymentFailed, PaymentResult, RemoteFailure, SendPayment}
-import fr.acinq.eclair.router.{Announcements, Data}
+import fr.acinq.eclair.router.{Announcements, Data, PublicChannel}
 import fr.acinq.eclair.wire.IncorrectOrUnknownPaymentDetails
 import fr.acinq.eclair.{MilliSatoshi, NodeParams, randomBytes32, secureRandom}
 
@@ -89,9 +89,10 @@ object Autoprobe {
 
   def pickPaymentDestination(nodeId: PublicKey, routingData: Data): Option[PublicKey] = {
     // we only pick direct peers with enabled public channels
-    val peers = routingData.updates
+    val peers = routingData.channels
       .collect {
-        case (desc, u) if desc.a == nodeId && Announcements.isEnabled(u.channelFlags) && routingData.channels.contains(u.shortChannelId) => desc.b // we only consider outgoing channels that are enabled and announced
+        case (shortChannelId, c@PublicChannel(ann, _, _, Some(u1), _))
+          if c.getNodeIdSameSideAs(u1) == nodeId && Announcements.isEnabled(u1.channelFlags) && routingData.channels.exists(_._1 == shortChannelId) => ann.nodeId2 // we only consider outgoing channels that are enabled and announced
       }
     if (peers.isEmpty) {
       None

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

@@ -23,6 +23,7 @@ import fr.acinq.eclair.router.Graph.GraphStructure.{DirectedGraph, GraphEdge}
 import fr.acinq.eclair.router.Router._
 import fr.acinq.eclair.wire.ChannelUpdate
 
+import scala.collection.immutable.SortedMap
 import scala.collection.mutable
 
 object Graph {
@@ -74,6 +75,7 @@ object Graph {
                         targetNode: PublicKey,
                         amount: MilliSatoshi,
                         ignoredEdges: Set[ChannelDesc],
+                        ignoredVertices: Set[PublicKey],
                         extraEdges: Set[GraphEdge],
                         pathsToFind: Int,
                         wr: Option[WeightRatios],
@@ -89,7 +91,7 @@ object Graph {
 
     // find the shortest path, k = 0
     val initialWeight = RichWeight(cost = amount, 0, CltvExpiryDelta(0), 0)
-    val shortestPath = dijkstraShortestPath(graph, sourceNode, targetNode, ignoredEdges, extraEdges, initialWeight, boundaries, currentBlockHeight, wr)
+    val shortestPath = dijkstraShortestPath(graph, sourceNode, targetNode, ignoredEdges, ignoredVertices, extraEdges, initialWeight, boundaries, currentBlockHeight, wr)
     shortestPaths += WeightedPath(shortestPath, pathWeight(shortestPath, amount, isPartial = false, currentBlockHeight, wr))
 
     // avoid returning a list with an empty path
@@ -125,7 +127,7 @@ object Graph {
           val returningEdges = rootPathEdges.lastOption.map(last => graph.getEdgesBetween(last.desc.b, last.desc.a)).toSeq.flatten.map(_.desc)
 
           // find the "spur" path, a sub-path going from the spur edge to the target avoiding previously found sub-paths
-          val spurPath = dijkstraShortestPath(graph, spurEdge.desc.a, targetNode, ignoredEdges ++ edgesToIgnore.toSet ++ returningEdges.toSet, extraEdges, rootPathWeight, boundaries, currentBlockHeight, wr)
+          val spurPath = dijkstraShortestPath(graph, spurEdge.desc.a, targetNode, ignoredEdges ++ edgesToIgnore.toSet ++ returningEdges.toSet, ignoredVertices, extraEdges, rootPathWeight, boundaries, currentBlockHeight, wr)
 
           // if there wasn't a path the spur will be empty
           if (spurPath.nonEmpty) {
@@ -178,6 +180,7 @@ object Graph {
                            sourceNode: PublicKey,
                            targetNode: PublicKey,
                            ignoredEdges: Set[ChannelDesc],
+                           ignoredVertices: Set[PublicKey],
                            extraEdges: Set[GraphEdge],
                            initialWeight: RichWeight,
                            boundaries: RichWeight => Boolean,
@@ -232,7 +235,7 @@ object Graph {
           if (edge.update.htlcMaximumMsat.forall(newMinimumKnownWeight.cost <= _) &&
             newMinimumKnownWeight.cost >= edge.update.htlcMinimumMsat &&
             boundaries(newMinimumKnownWeight) && // check if this neighbor edge would break off the 'boundaries'
-            !ignoredEdges.contains(edge.desc)
+            !ignoredEdges.contains(edge.desc) && !ignoredVertices.contains(neighbor)
           ) {
 
             // we call containsKey first because "getOrDefault" is not available in JDK7
@@ -533,21 +536,32 @@ object Graph {
       def apply(edge: GraphEdge): DirectedGraph = new DirectedGraph(Map()).addEdge(edge.desc, edge.update)
 
       def apply(edges: Seq[GraphEdge]): DirectedGraph = {
-        makeGraph(edges.map(e => e.desc -> e.update).toMap)
+        DirectedGraph().addEdges(edges.map(e => (e.desc, e.update)))
       }
 
       // optimized constructor
-      def makeGraph(descAndUpdates: Map[ChannelDesc, ChannelUpdate]): DirectedGraph = {
+      def makeGraph(channels: SortedMap[ShortChannelId, PublicChannel]): DirectedGraph = {
 
         // initialize the map with the appropriate size to avoid resizing during the graph initialization
         val mutableMap = new {} with mutable.HashMap[PublicKey, List[GraphEdge]] {
-          override def initialSize: Int = descAndUpdates.size + 1
+          override def initialSize: Int = channels.size + 1
         }
 
         // add all the vertices and edges in one go
-        descAndUpdates.foreach { case (desc, update) =>
-          // create or update vertex (desc.b) and update its neighbor
-          mutableMap.put(desc.b, GraphEdge(desc, update) +: mutableMap.getOrElse(desc.b, List.empty[GraphEdge]))
+        channels.values.foreach { channel =>
+          channel.update_1_opt.foreach { u1 =>
+            val desc1 = Router.getDesc(u1, channel.ann)
+            addDescToMap(desc1, u1)
+          }
+
+          channel.update_2_opt.foreach { u2 =>
+            val desc2 = Router.getDesc(u2, channel.ann)
+            addDescToMap(desc2, u2)
+          }
+        }
+
+        def addDescToMap(desc: ChannelDesc, u: ChannelUpdate) = {
+          mutableMap.put(desc.b, GraphEdge(desc, u) +: mutableMap.getOrElse(desc.b, List.empty[GraphEdge]))
           mutableMap.get(desc.a) match {
             case None => mutableMap += desc.a -> List.empty[GraphEdge]
             case _ =>