Trampoline/MPP API changes

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

@@ -263,7 +263,7 @@ class SqliteAuditDb(sqlite: Connection) extends AuditDb with Logging {
     }
 
   override def listRelayed(from: Long, to: Long): Seq[PaymentRelayed] =
-    using(sqlite.prepareStatement("SELECT * FROM relayed WHERE timestamp >= ? AND timestamp < ?")) { statement =>
+    using(sqlite.prepareStatement("SELECT * FROM relayed WHERE timestamp >= ? AND timestamp < ? ORDER BY timestamp")) { statement =>
       statement.setLong(1, from)
       statement.setLong(2, to)
       val rs = statement.executeQuery()
@@ -278,17 +278,18 @@ class SqliteAuditDb(sqlite: Connection) extends AuditDb with Logging {
           rs.getLong("timestamp"))
         relayedByHash = relayedByHash + (paymentHash -> (relayedByHash.getOrElse(paymentHash, Nil) :+ part))
       }
-      relayedByHash.map {
-        case (paymentHash, parts) => parts.head.relayType match {
-          case "channel" => parts.foldLeft(ChannelPaymentRelayed(0 msat, 0 msat, paymentHash, ByteVector32.Zeroes, ByteVector32.Zeroes, parts.head.timestamp)) {
-            case (e, part) if part.direction == "IN" => e.copy(amountIn = part.amount, fromChannelId = part.channelId)
-            case (e, part) if part.direction == "OUT" => e.copy(amountOut = part.amount, toChannelId = part.channelId)
+      relayedByHash.flatMap {
+        case (paymentHash, parts) =>
+          val timestamp = parts.head.timestamp
+          // We may have been routing multiple payments for the same payment_hash (MPP) in both cases (trampoline and channel).
+          val incoming = parts.filter(_.direction == "IN").map(p => PaymentRelayed.Part(p.amount, p.channelId))
+          val outgoing = parts.filter(_.direction == "OUT").map(p => PaymentRelayed.Part(p.amount, p.channelId))
+          parts.head.relayType match {
+            case "channel" => incoming.zip(outgoing).map {
+              case (in, out) => ChannelPaymentRelayed(in.amount, out.amount, paymentHash, in.channelId, out.channelId, timestamp)
+            }
+            case "trampoline" => TrampolinePaymentRelayed(paymentHash, incoming, outgoing, timestamp) :: Nil
           }
-          case "trampoline" => parts.foldLeft(TrampolinePaymentRelayed(paymentHash, Nil, Nil, parts.head.timestamp)) {
-            case (e, part) if part.direction == "IN" => e.copy(incoming = e.incoming :+ PaymentRelayed.Part(part.amount, part.channelId))
-            case (e, part) if part.direction == "OUT" => e.copy(outgoing = e.outgoing :+ PaymentRelayed.Part(part.amount, part.channelId))
-          }
-        }
       }.toSeq.sortBy(_.timestamp)
     }
 

eclair-core/src/test/scala/fr/acinq/eclair/db/SqliteAuditDbSpec.scala

@@ -62,6 +62,9 @@ class SqliteAuditDbSpec extends FunSuite {
     val e9 = ChannelErrorOccurred(null, randomBytes32, randomKey.publicKey, null, LocalError(new RuntimeException("oops")), isFatal = true)
     val e10 = ChannelErrorOccurred(null, randomBytes32, randomKey.publicKey, null, RemoteError(wire.Error(randomBytes32, "remote oops")), isFatal = true)
     val e11 = TrampolinePaymentRelayed(randomBytes32, Seq(PaymentRelayed.Part(20000 msat, randomBytes32), PaymentRelayed.Part(22000 msat, randomBytes32)), Seq(PaymentRelayed.Part(20000 msat, randomBytes32), PaymentRelayed.Part(10000 msat, randomBytes32), PaymentRelayed.Part(10000 msat, randomBytes32)))
+    val multiPartPaymentHash = randomBytes32
+    val e12 = ChannelPaymentRelayed(13000 msat, 11000 msat, multiPartPaymentHash, randomBytes32, randomBytes32)
+    val e13 = ChannelPaymentRelayed(15000 msat, 12500 msat, multiPartPaymentHash, randomBytes32, randomBytes32)
 
     db.add(e1)
     db.add(e2)
@@ -74,11 +77,13 @@ class SqliteAuditDbSpec extends FunSuite {
     db.add(e9)
     db.add(e10)
     db.add(e11)
+    db.add(e12)
+    db.add(e13)
 
     assert(db.listSent(from = 0L, to = (Platform.currentTime.milliseconds + 15.minute).toMillis).toSet === Set(e1, e5, e6))
     assert(db.listSent(from = 100000L, to = (Platform.currentTime.milliseconds + 1.minute).toMillis).toList === List(e1))
     assert(db.listReceived(from = 0L, to = (Platform.currentTime.milliseconds + 1.minute).toMillis).toList === List(e2))
-    assert(db.listRelayed(from = 0L, to = (Platform.currentTime.milliseconds + 1.minute).toMillis).toList === List(e3, e11))
+    assert(db.listRelayed(from = 0L, to = (Platform.currentTime.milliseconds + 1.minute).toMillis).toList === List(e3, e11, e12, e13))
     assert(db.listNetworkFees(from = 0L, to = (Platform.currentTime.milliseconds + 1.minute).toMillis).size === 1)
     assert(db.listNetworkFees(from = 0L, to = (Platform.currentTime.milliseconds + 1.minute).toMillis).head.txType === "mutual")
   }
@@ -351,7 +356,7 @@ class SqliteAuditDbSpec extends FunSuite {
     assert(postMigrationDb.listRelayed(100, 160) === Seq(relayed1, relayed2, relayed3))
   }
 
-  test("fails if the DB contains invalid values") {
+  test("handle invalid values in the DB") {
     val sqlite = TestConstants.sqliteInMemory()
     val db = new SqliteAuditDb(sqlite)
 
@@ -360,7 +365,7 @@ class SqliteAuditDbSpec extends FunSuite {
       statement.setLong(2, 42)
       statement.setBytes(3, randomBytes32.toArray)
       statement.setString(4, "IN")
-      statement.setString(5, "unknown") // invalid relay type
+      statement.setString(5, "unknown") // invalid relay type -> will throw
       statement.setLong(6, 10)
       statement.executeUpdate()
     }
@@ -371,13 +376,13 @@ class SqliteAuditDbSpec extends FunSuite {
       statement.setBytes(1, randomBytes32.toArray)
       statement.setLong(2, 51)
       statement.setBytes(3, randomBytes32.toArray)
-      statement.setString(4, "UP") // invalid direction
+      statement.setString(4, "UP") // invalid direction -> is simply ignored
       statement.setString(5, "channel")
       statement.setLong(6, 20)
       statement.executeUpdate()
     }
 
-    assertThrows[MatchError](db.listRelayed(15, 25))
+    assert(db.listRelayed(15, 25) === Nil)
 
     val paymentHash = randomBytes32
     val channelId = randomBytes32
@@ -386,13 +391,13 @@ class SqliteAuditDbSpec extends FunSuite {
       statement.setBytes(1, paymentHash.toArray)
       statement.setLong(2, 65)
       statement.setBytes(3, channelId.toArray)
-      statement.setString(4, "IN")
+      statement.setString(4, "IN") // missing a corresponding OUT -> is simply ignored
       statement.setString(5, "channel")
       statement.setLong(6, 30)
       statement.executeUpdate()
     }
 
-    assert(db.listRelayed(25, 35) === Seq(ChannelPaymentRelayed(65 msat, 0 msat, paymentHash, channelId, ByteVector32.Zeroes, 30)))
+    assert(db.listRelayed(25, 35) === Nil)
   }
 
 }