Use lower endorsement payments to increase confidence of higher endorsement payments

thomash-acinq · thomash-acinq · commit 054834824ac1 · 2025-05-28T11:21:59.000+02:00
diff --git a/eclair-core/src/main/scala/fr/acinq/eclair/reputation/Reputation.scala b/eclair-core/src/main/scala/fr/acinq/eclair/reputation/Reputation.scala
@@ -1,5 +1,5 @@
 /*
- * Copyright 2023 ACINQ SAS
+ * Copyright 2025 ACINQ SAS
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -17,76 +17,95 @@
 package fr.acinq.eclair.reputation
 
 import fr.acinq.bitcoin.scalacompat.ByteVector32
-import fr.acinq.eclair.reputation.Reputation.HtlcId
 import fr.acinq.eclair.{MilliSatoshi, TimestampMilli}
 
+import scala.collection.mutable
 import scala.concurrent.duration.FiniteDuration
 
 /**
- * Created by thomash on 21/07/2023.
+ *
+ * @param weight           How much fees we would have collected in the past if all payments had succeeded (exponential moving average).
+ * @param score            How much fees we have collected in the past (exponential moving average).
+ * @param lastSettlementAt Timestamp of the last recorded payment settlement.
  */
+case class PastScore(weight: Double, score: Double, lastSettlementAt: TimestampMilli)
+
+/** We're relaying that payment and are waiting for it to settle. */
+case class PendingPayment(fee: MilliSatoshi, endorsement: Int, startedAt: TimestampMilli) {
+  def weight(now: TimestampMilli, minDuration: FiniteDuration, multiplier: Double): Double = {
+    val duration = now - startedAt
+    fee.toLong.toDouble * (duration / minDuration).max(multiplier)
+  }
+}
+
+case class HtlcId(channelId: ByteVector32, id: Long)
 
 /**
- * Local reputation for a given incoming node, that should be tracked for each incoming endorsement level.
+ * Local reputation for a given node.
  *
- * @param pastWeight        How much fees we would have collected in the past if all payments had succeeded (exponential moving average).
- * @param pastScore         How much fees we have collected in the past (exponential moving average).
- * @param lastSettlementAt  Timestamp of the last recorded payment settlement.
  * @param pending           Set of pending payments (payments may contain multiple HTLCs when using trampoline).
  * @param halfLife          Half life for the exponential moving average.
  * @param maxRelayDuration  Duration after which payments are penalized for staying pending too long.
  * @param pendingMultiplier How much to penalize pending payments.
  */
-case class Reputation(pastWeight: Double, pastScore: Double, lastSettlementAt: TimestampMilli, pending: Map[HtlcId, Reputation.PendingPayment], halfLife: FiniteDuration, maxRelayDuration: FiniteDuration, pendingMultiplier: Double) {
-  private def decay(now: TimestampMilli): Double = scala.math.pow(0.5, (now - lastSettlementAt) / halfLife)
-
-  private def pendingWeight(now: TimestampMilli): Double = pending.values.map(_.weight(now, maxRelayDuration, pendingMultiplier)).sum
+case class Reputation(pastScores: Array[PastScore], pending: mutable.Map[HtlcId, PendingPayment], halfLife: FiniteDuration, maxRelayDuration: FiniteDuration, pendingMultiplier: Double) {
+  private def decay(now: TimestampMilli, lastSettlementAt: TimestampMilli): Double = scala.math.pow(0.5, (now - lastSettlementAt) / halfLife)
 
   /**
    * Estimate the confidence that a payment will succeed.
    */
-  def getConfidence(fee: MilliSatoshi, now: TimestampMilli = TimestampMilli.now()): Double = {
-    val d = decay(now)
-    d * pastScore / (d * pastWeight + pendingWeight(now) + fee.toLong.toDouble * pendingMultiplier)
+  def getConfidence(fee: MilliSatoshi, endorsement: Int, now: TimestampMilli = TimestampMilli.now()): Double = {
+    val weights = Array.fill(Reputation.endorsementLevels)(0.0)
+    val scores = Array.fill(Reputation.endorsementLevels)(0.0)
+    for (e <- 0 until Reputation.endorsementLevels) {
+      val d = decay(now, pastScores(e).lastSettlementAt)
+      weights(e) += d * pastScores(e).weight
+      scores(e) += d * pastScores(e).score
+    }
+    for (p <- pending.values) {
+      weights(p.endorsement) += p.weight(now, maxRelayDuration, pendingMultiplier)
+    }
+    weights(endorsement) += fee.toLong.toDouble * pendingMultiplier
+    /*
+     Higher endorsement buckets may have fewer payments which makes the weight of pending payments disproportionately
+     important. To counter this effect, we try adding payments from the lower buckets to see if it gives us a higher
+     confidence score.
+     It is acceptable to use payments with lower endorsements to increase the confidence score but not payments with
+     higher endorsements.
+    */
+    var score = scores(endorsement)
+    var weight = weights(endorsement)
+    var confidence = score / weight
+    for (e <- Range.inclusive(endorsement - 1, 0, step = -1)) {
+      score += scores(e)
+      weight += weights(e)
+      confidence = confidence.max(score / weight)
+    }
+    confidence
   }
 
   /**
    * Register a pending relay.
-   *
-   * @return updated reputation
    */
-  def attempt(htlcId: HtlcId, fee: MilliSatoshi, now: TimestampMilli = TimestampMilli.now()): Reputation =
-    copy(pending = pending + (htlcId -> Reputation.PendingPayment(fee, now)))
+  def attempt(htlcId: HtlcId, fee: MilliSatoshi, endorsement: Int, now: TimestampMilli = TimestampMilli.now()): Unit =
+    pending(htlcId) = PendingPayment(fee, endorsement, now)
 
   /**
    * When a payment is settled, we record whether it succeeded and how long it took.
-   *
-   * @return updated reputation
    */
-  def record(htlcId: HtlcId, isSuccess: Boolean, now: TimestampMilli = TimestampMilli.now()): Reputation = {
-    pending.get(htlcId) match {
-      case Some(p) =>
-        val d = decay(now)
-        val newWeight = d * pastWeight + p.weight(now, maxRelayDuration, if (isSuccess) 1.0 else 0.0)
-        val newScore = d * pastScore + (if (isSuccess) p.fee.toLong.toDouble else 0)
-        Reputation(newWeight, newScore, now, pending - htlcId, halfLife, maxRelayDuration, pendingMultiplier)
-      case None => this
-    }
-  }
+  def record(htlcId: HtlcId, isSuccess: Boolean, now: TimestampMilli = TimestampMilli.now()): Unit =
+    pending.remove(htlcId).foreach(p => {
+      val d = decay(now, pastScores(p.endorsement).lastSettlementAt)
+      val newWeight = d * pastScores(p.endorsement).weight + p.weight(now, maxRelayDuration, if (isSuccess) 1.0 else 0.0)
+      val newScore = d * pastScores(p.endorsement).score + (if (isSuccess) p.fee.toLong.toDouble else 0)
+      pastScores(p.endorsement) = PastScore(newWeight, newScore, now)
+    })
 }
 
 object Reputation {
-  case class HtlcId(channelId: ByteVector32, id: Long)
-
-  /** We're relaying that payment and are waiting for it to settle. */
-  case class PendingPayment(fee: MilliSatoshi, startedAt: TimestampMilli) {
-    def weight(now: TimestampMilli, minDuration: FiniteDuration, multiplier: Double): Double = {
-      val duration = now - startedAt
-      fee.toLong.toDouble * (duration / minDuration).max(multiplier)
-    }
-  }
+  val endorsementLevels = 8
 
   case class Config(enabled: Boolean, halfLife: FiniteDuration, maxRelayDuration: FiniteDuration, pendingMultiplier: Double)
 
-  def init(config: Config): Reputation = Reputation(0.0, 0.0, TimestampMilli.min, Map.empty, config.halfLife, config.maxRelayDuration, config.pendingMultiplier)
+  def init(config: Config): Reputation = Reputation(Array.fill(endorsementLevels)(PastScore(0.0, 0.0, TimestampMilli.min)), mutable.HashMap.empty, config.halfLife, config.maxRelayDuration, config.pendingMultiplier)
 }
diff --git a/eclair-core/src/main/scala/fr/acinq/eclair/reputation/ReputationRecorder.scala b/eclair-core/src/main/scala/fr/acinq/eclair/reputation/ReputationRecorder.scala
@@ -1,5 +1,5 @@
 /*
- * Copyright 2023 ACINQ SAS
+ * Copyright 2025 ACINQ SAS
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -23,17 +23,11 @@ import fr.acinq.bitcoin.scalacompat.Crypto.PublicKey
 import fr.acinq.eclair.MilliSatoshi
 import fr.acinq.eclair.channel.Upstream.Hot
 import fr.acinq.eclair.channel.{OutgoingHtlcAdded, OutgoingHtlcFailed, OutgoingHtlcFulfilled, Upstream}
-import fr.acinq.eclair.reputation.Reputation.HtlcId
 import fr.acinq.eclair.wire.protocol.{UpdateFailHtlc, UpdateFailMalformedHtlc}
 import ReputationRecorder._
 
 import scala.collection.mutable
 
-
-/**
- * Created by thomash on 21/07/2023.
- */
-
 object ReputationRecorder {
   // @formatter:off
   sealed trait Command
@@ -44,16 +38,6 @@ object ReputationRecorder {
   private case class WrappedOutgoingHtlcFulfilled(fulfilled: OutgoingHtlcFulfilled) extends Command
   // @formatter:on
 
-  /**
-   * @param nodeId      nodeId of the upstream peer.
-   * @param endorsement endorsement value set by the upstream peer in the HTLC we received.
-   */
-  case class PeerEndorsement(nodeId: PublicKey, endorsement: Int)
-
-  object PeerEndorsement {
-    def apply(channel: Upstream.Hot.Channel): PeerEndorsement = PeerEndorsement(channel.receivedFrom, channel.add.endorsement)
-  }
-
   /** Confidence that the outgoing HTLC will succeed. */
   case class Confidence(value: Double)
 
@@ -67,24 +51,37 @@ object ReputationRecorder {
 }
 
 class ReputationRecorder(config: Reputation.Config, context: ActorContext[ReputationRecorder.Command]) {
-  private val reputations: mutable.Map[PeerEndorsement, Reputation] = mutable.HashMap.empty.withDefaultValue(Reputation.init(config))
+  private val reputations: mutable.Map[PublicKey, Reputation] = mutable.HashMap.empty.withDefault(_ => Reputation.init(config))
   private val pending: mutable.Map[HtlcId, Upstream.Hot] = mutable.HashMap.empty
 
+  private def getReputation(nodeId: PublicKey): Reputation = {
+    reputations.get(nodeId) match {
+      case Some(reputation) => reputation
+      case None => {
+        val reputation = Reputation.init(config)
+        reputations(nodeId) = reputation
+        reputation
+      }
+    }
+  }
+
   def run(): Behavior[Command] =
     Behaviors.receiveMessage {
       case GetConfidence(replyTo, upstream, fee) =>
-        val confidence = reputations(PeerEndorsement(upstream)).getConfidence(fee)
+        val confidence = reputations(upstream.receivedFrom).getConfidence(fee, upstream.add.endorsement)
         replyTo ! Confidence(confidence)
         Behaviors.same
 
       case GetTrampolineConfidence(replyTo, upstream, totalFee) =>
         val confidence =
           upstream.received
-            .groupMapReduce(r => PeerEndorsement(r.receivedFrom, r.add.endorsement))(_.add.amountMsat)(_ + _)
+            .groupMapReduce(_.receivedFrom)(r => (r.add.amountMsat, r.add.endorsement)){
+              case ((amount1, endorsement1), (amount2, endorsement2)) => (amount1 + amount2, endorsement1 min endorsement2)
+            }
             .map {
-              case (peerEndorsement, amount) =>
+              case (nodeId, (amount, endorsement)) =>
                 val fee = amount * totalFee.toLong / upstream.amountIn.toLong
-                reputations(peerEndorsement).getConfidence(fee)
+                reputations(nodeId).getConfidence(fee, endorsement)
             }
             .min
         replyTo ! Confidence(confidence)
@@ -94,11 +91,11 @@ class ReputationRecorder(config: Reputation.Config, context: ActorContext[Reputa
         val htlcId = HtlcId(add.channelId, add.id)
         upstream match {
           case channel: Hot.Channel =>
-            reputations(PeerEndorsement(channel)) = reputations(PeerEndorsement(channel)).attempt(htlcId, fee)
+            getReputation(channel.receivedFrom).attempt(htlcId, fee, channel.add.endorsement)
             pending += (htlcId -> upstream)
           case trampoline: Hot.Trampoline =>
             trampoline.received.foreach(channel =>
-              reputations(PeerEndorsement(channel)) = reputations(PeerEndorsement(channel)).attempt(htlcId, fee * channel.amountIn.toLong / trampoline.amountIn.toLong)
+              getReputation(channel.receivedFrom).attempt(htlcId, fee * channel.amountIn.toLong / trampoline.amountIn.toLong, channel.add.endorsement)
             )
             pending += (htlcId -> upstream)
           case _: Upstream.Local => ()
@@ -112,10 +109,10 @@ class ReputationRecorder(config: Reputation.Config, context: ActorContext[Reputa
         }
         pending.get(htlcId) match {
           case Some(channel: Hot.Channel) =>
-            reputations(PeerEndorsement(channel)) = reputations(PeerEndorsement(channel)).record(htlcId, isSuccess = false)
+            getReputation(channel.receivedFrom).record(htlcId, isSuccess = false)
           case Some(trampoline: Hot.Trampoline) =>
             trampoline.received.foreach(channel =>
-              reputations(PeerEndorsement(channel)) = reputations(PeerEndorsement(channel)).record(htlcId, isSuccess = false)
+              getReputation(channel.receivedFrom).record(htlcId, isSuccess = false)
             )
           case _ => ()
         }
@@ -126,10 +123,10 @@ class ReputationRecorder(config: Reputation.Config, context: ActorContext[Reputa
         val htlcId = HtlcId(fulfill.channelId, fulfill.id)
         pending.get(htlcId) match {
           case Some(channel: Hot.Channel) =>
-            reputations(PeerEndorsement(channel)) = reputations(PeerEndorsement(channel)).record(htlcId, isSuccess = true)
+            getReputation(channel.receivedFrom).record(htlcId, isSuccess = true)
           case Some(trampoline: Hot.Trampoline) =>
             trampoline.received.foreach(channel =>
-              reputations(PeerEndorsement(channel)) = reputations(PeerEndorsement(channel)).record(htlcId, isSuccess = true)
+              getReputation(channel.receivedFrom).record(htlcId, isSuccess = true)
             )
           case _ => ()
         }
diff --git a/eclair-core/src/test/scala/fr/acinq/eclair/reputation/ReputationSpec.scala b/eclair-core/src/test/scala/fr/acinq/eclair/reputation/ReputationSpec.scala
@@ -27,48 +27,48 @@ class ReputationSpec extends AnyFunSuite {
   val (htlcId1, htlcId2, htlcId3, htlcId4, htlcId5, htlcId6, htlcId7) = (HtlcId(randomBytes32(), 1), HtlcId(randomBytes32(), 2), HtlcId(randomBytes32(), 3), HtlcId(randomBytes32(), 4), HtlcId(randomBytes32(), 5), HtlcId(randomBytes32(), 6), HtlcId(randomBytes32(), 7))
 
   test("basic") {
-    val r0 = Reputation.init(Config(enabled = true, 1 day, 1 second, 2))
-    assert(r0.getConfidence(10000 msat) == 0)
-    val r1 = r0.attempt(htlcId1, 10000 msat)
-    val r2 = r1.record(htlcId1, isSuccess = true)
-    val r3 = r2.attempt(htlcId2, 10000 msat)
-    assert(r2.getConfidence(10000 msat) === (1.0 / 3) +- 0.001)
-    val r4 = r3.attempt(htlcId3, 10000 msat)
-    assert(r3.getConfidence(10000 msat) === (1.0 / 5) +- 0.001)
-    val r5 = r4.record(htlcId2, isSuccess = true)
-    val r6 = r5.record(htlcId3, isSuccess = true)
-    val r7 = r6.attempt(htlcId4, 1 msat)
-    assert(r6.getConfidence(1 msat) === 1.0 +- 0.001)
-    val r8 = r7.attempt(htlcId5, 40000 msat)
-    assert(r7.getConfidence(40000 msat) === (3.0 / 11) +- 0.001)
-    val r9 = r8.attempt(htlcId6, 10000 msat)
-    assert(r8.getConfidence(10000 msat) === (3.0 / 13) +- 0.001)
-    val r10 = r9.record(htlcId6, isSuccess = false)
-    assert(r10.getConfidence(10000 msat) === (3.0 / 13) +- 0.001)
+    val r = Reputation.init(Config(enabled = true, 1 day, 1 second, 2))
+    assert(r.getConfidence(10000 msat, 0) == 0)
+    r.attempt(htlcId1, 10000 msat, 0)
+    r.record(htlcId1, isSuccess = true)
+    r.attempt(htlcId2, 10000 msat, 0)
+    assert(r.getConfidence(10000 msat, 0) === (1.0 / 3) +- 0.001)
+    r.attempt(htlcId3, 10000 msat, 0)
+    assert(r.getConfidence(10000 msat, 0) === (1.0 / 5) +- 0.001)
+    r.record(htlcId2, isSuccess = true)
+    r.record(htlcId3, isSuccess = true)
+    r.attempt(htlcId4, 1 msat, 0)
+    assert(r.getConfidence(1 msat, 0) === 1.0 +- 0.001)
+    r.attempt(htlcId5, 40000 msat, 0)
+    assert(r.getConfidence(40000 msat, 0) === (3.0 / 11) +- 0.001)
+    r.attempt(htlcId6, 10000 msat, 0)
+    assert(r.getConfidence(10000 msat, 0) === (3.0 / 13) +- 0.001)
+    r.record(htlcId6, isSuccess = false)
+    assert(r.getConfidence(10000 msat, 0) === (3.0 / 13) +- 0.001)
   }
 
   test("long HTLC") {
-    val r0 = Reputation.init(Config(enabled = true, 1000 day, 1 second, 10))
-    assert(r0.getConfidence(100000 msat, TimestampMilli(0)) == 0)
-    val r1 = r0.attempt(htlcId1, 100000 msat, TimestampMilli(0))
-    val r2 = r1.record(htlcId1, isSuccess = true, now = TimestampMilli(0))
-    assert(r2.getConfidence(1000 msat, TimestampMilli(0)) === (10.0 / 11) +- 0.001)
-    val r3 = r2.attempt(htlcId2, 1000 msat, TimestampMilli(0))
-    val r4 = r3.record(htlcId2, isSuccess = false, now = TimestampMilli(0) + 100.seconds)
-    assert(r4.getConfidence(0 msat, now = TimestampMilli(0) + 100.seconds) === 0.5 +- 0.001)
+    val r = Reputation.init(Config(enabled = true, 1000 day, 1 second, 10))
+    assert(r.getConfidence(100000 msat, 0, TimestampMilli(0)) == 0)
+    r.attempt(htlcId1, 100000 msat, 0, TimestampMilli(0))
+    r.record(htlcId1, isSuccess = true, now = TimestampMilli(0))
+    assert(r.getConfidence(1000 msat, 0, TimestampMilli(0)) === (10.0 / 11) +- 0.001)
+    r.attempt(htlcId2, 1000 msat, 0, TimestampMilli(0))
+    r.record(htlcId2, isSuccess = false, now = TimestampMilli(0) + 100.seconds)
+    assert(r.getConfidence(0 msat, 0, now = TimestampMilli(0) + 100.seconds) === 0.5 +- 0.001)
   }
 
   test("exponential decay") {
-    val r0 = Reputation.init(Config(enabled = true, 100 seconds, 1 second, 1))
-    val r1 = r0.attempt(htlcId1, 1000 msat, TimestampMilli(0))
-    val r2 = r1.record(htlcId1, isSuccess = true, now = TimestampMilli(0))
-    assert(r2.getConfidence(1000 msat, TimestampMilli(0)) == 1.0 / 2)
-    val r3 = r2.attempt(htlcId2, 1000 msat, TimestampMilli(0))
-    val r4 = r3.record(htlcId2, isSuccess = true, now = TimestampMilli(0))
-    assert(r4.getConfidence(1000 msat, TimestampMilli(0)) == 2.0 / 3)
-    val r5 = r4.attempt(htlcId3, 1000 msat, TimestampMilli(0))
-    val r6 = r5.record(htlcId3, isSuccess = true, now = TimestampMilli(0))
-    assert(r6.getConfidence(1000 msat, TimestampMilli(0) + 100.seconds) == 1.5 / 2.5)
-    assert(r6.getConfidence(1000 msat, TimestampMilli(0) + 1.hour) < 0.000001)
+    val r = Reputation.init(Config(enabled = true, 100 seconds, 1 second, 1))
+    r.attempt(htlcId1, 1000 msat, 0, TimestampMilli(0))
+    r.record(htlcId1, isSuccess = true, now = TimestampMilli(0))
+    assert(r.getConfidence(1000 msat, 0, TimestampMilli(0)) == 1.0 / 2)
+    r.attempt(htlcId2, 1000 msat, 0, TimestampMilli(0))
+    r.record(htlcId2, isSuccess = true, now = TimestampMilli(0))
+    assert(r.getConfidence(1000 msat, 0, TimestampMilli(0)) == 2.0 / 3)
+    r.attempt(htlcId3, 1000 msat, 0, TimestampMilli(0))
+    r.record(htlcId3, isSuccess = true, now = TimestampMilli(0))
+    assert(r.getConfidence(1000 msat, 0, TimestampMilli(0) + 100.seconds) == 1.5 / 2.5)
+    assert(r.getConfidence(1000 msat, 0, TimestampMilli(0) + 1.hour) < 0.000001)
   }
 }
