Polish

Author: thomash-acinq

docs/release-notes/eclair-vnext.md

@@ -43,12 +43,12 @@ eclair.relay.peer-reputation {
     // value, as described by https://github.com/lightning/blips/blob/master/blip-0004.md,
     enabled = true
     // Reputation decays with the following half life to emphasize recent behavior.
-    half-life = 7 days
+    half-life = 15 days
     // Payments that stay pending for longer than this get penalized
     max-relay-duration = 12 seconds
     // Pending payments are counted as failed, and because they could potentially stay pending for a very long time,
     // the following multiplier is applied.
-    pending-multiplier = 1000 // A pending payment counts as a thousand failed ones.
+    pending-multiplier = 200 // A pending payment counts as a thousand failed ones.
 }
 ```
 

eclair-core/src/main/scala/fr/acinq/eclair/channel/ChannelData.scala

@@ -147,19 +147,25 @@ case class INPUT_RESTORED(data: PersistentChannelData)
 sealed trait Upstream { def amountIn: MilliSatoshi }
 object Upstream {
   /** We haven't restarted and have full information about the upstream parent(s). */
-  sealed trait Hot extends Upstream
+  sealed trait Hot extends Upstream {
+    def show: String
+  }
   object Hot {
     /** Our node is forwarding a single incoming HTLC. */
     case class Channel(add: UpdateAddHtlc, receivedAt: TimestampMilli, receivedFrom: PublicKey) extends Hot {
       override val amountIn: MilliSatoshi = add.amountMsat
       val expiryIn: CltvExpiry = add.cltvExpiry
+
+      override def show: String = s"Channel(receivedAt=${receivedAt.toLong}, receivedFrom=${receivedFrom.toHex}, endorsement=${add.endorsement})"
     }
     /** Our node is forwarding a payment based on a set of HTLCs from potentially multiple upstream channels. */
     case class Trampoline(received: List[Channel]) extends Hot {
       override val amountIn: MilliSatoshi = received.map(_.add.amountMsat).sum
       // We must use the lowest expiry of the incoming HTLC set.
       val expiryIn: CltvExpiry = received.map(_.add.cltvExpiry).min
       val receivedAt: TimestampMilli = received.map(_.receivedAt).max
+
+      override def show: String = s"Trampoline(${received.map(_.show).mkString(",")})"
     }
   }
 
@@ -183,7 +189,11 @@ object Upstream {
   }
 
   /** Our node is the origin of the payment: there are no matching upstream HTLCs. */
-  case class Local(id: UUID) extends Hot with Cold { override val amountIn: MilliSatoshi = 0 msat }
+  case class Local(id: UUID) extends Hot with Cold {
+    override val amountIn: MilliSatoshi = 0 msat
+
+    override def show: String = toString
+  }
 }
 
 /**

eclair-core/src/main/scala/fr/acinq/eclair/channel/Commitments.scala

@@ -537,7 +537,7 @@ case class Commitment(fundingTxIndex: Long,
       }
     }
     val occupancy = (outgoingHtlcs.size.toDouble / maxAcceptedHtlcs).max(htlcValueInFlight.toLong.toDouble / allowedHtlcValueInFlight.toBigInt.toDouble)
-    if (confidence + 0.05 < occupancy) {
+    if (confidence + 0.1 < occupancy) { // We add a 10% tolerance to enable payments from nodes without history and to account for the fact that even at the highest endorsement level we still expect a confidence of less than 93.75%.
       return Left(ConfidenceTooLow(params.channelId, confidence, occupancy))
     }
 

eclair-core/src/main/scala/fr/acinq/eclair/channel/fsm/Channel.scala

@@ -514,7 +514,9 @@ class Channel(val nodeParams: NodeParams, val channelKeys: ChannelKeys, val wall
         case Right((commitments1, add)) =>
           if (c.commit) self ! CMD_SIGN()
           context.system.eventStream.publish(AvailableBalanceChanged(self, d.channelId, d.aliases, commitments1, d.lastAnnouncement_opt))
-          context.system.eventStream.publish(OutgoingHtlcAdded(add, remoteNodeId, c.origin.upstream, nodeFee(d.channelUpdate.relayFees, add.amountMsat)))
+          val relayFee = nodeFee(d.channelUpdate.relayFees, add.amountMsat)
+          context.system.eventStream.publish(OutgoingHtlcAdded(add, remoteNodeId, c.origin.upstream, relayFee))
+          log.info("OutgoingHtlcAdded: channelId={}, id={}, endorsement={}, remoteNodeId={}, upstream={}, fee={}", Array(add.channelId.toHex, add.id, add.endorsement, remoteNodeId.toHex, c.origin.upstream.show, relayFee))
           handleCommandSuccess(c, d.copy(commitments = commitments1)) sending add
         case Left(cause) => handleAddHtlcCommandError(c, cause, Some(d.channelUpdate))
       }
@@ -542,6 +544,7 @@ class Channel(val nodeParams: NodeParams, val channelKeys: ChannelKeys, val wall
           // we forward preimages as soon as possible to the upstream channel because it allows us to pull funds
           relayer ! RES_ADD_SETTLED(origin, htlc, HtlcResult.RemoteFulfill(fulfill))
           context.system.eventStream.publish(OutgoingHtlcFulfilled(fulfill))
+          log.info("OutgoingHtlcFulfilled: channelId={}, id={}", fulfill.channelId.toHex, fulfill.id)
           stay() using d.copy(commitments = commitments1)
         case Left(cause) => handleLocalError(cause, d, Some(fulfill))
       }
@@ -576,13 +579,15 @@ class Channel(val nodeParams: NodeParams, val channelKeys: ChannelKeys, val wall
 
     case Event(fail: UpdateFailHtlc, d: DATA_NORMAL) =>
       context.system.eventStream.publish(OutgoingHtlcFailed(fail))
+      log.info("OutgoingHtlcFailed: channelId={}, id={}", fail.channelId.toHex, fail.id)
       d.commitments.receiveFail(fail) match {
         case Right((commitments1, _, _)) => stay() using d.copy(commitments = commitments1)
         case Left(cause) => handleLocalError(cause, d, Some(fail))
       }
 
     case Event(fail: UpdateFailMalformedHtlc, d: DATA_NORMAL) =>
       context.system.eventStream.publish(OutgoingHtlcFailed(fail))
+      log.info("OutgoingHtlcFailed: channelId={}, id={}", fail.channelId.toHex, fail.id)
       d.commitments.receiveFailMalformed(fail) match {
         case Right((commitments1, _, _)) => stay() using d.copy(commitments = commitments1)
         case Left(cause) => handleLocalError(cause, d, Some(fail))

eclair-core/src/main/scala/fr/acinq/eclair/reputation/Reputation.scala

@@ -23,15 +23,16 @@ import scala.collection.mutable
 import scala.concurrent.duration.FiniteDuration
 
 /**
+ * Reputation score per endorsement level.
  *
- * @param weight           How much fees we would have collected in the past if all payments had succeeded (exponential moving average).
+ * @param weight           How much fees we would have collected in the past if all HTLCs 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.
+ * @param lastSettlementAt Timestamp of the last recorded HTLC 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) {
+/** We're relaying that HTLC and are waiting for it to settle. */
+case class PendingHtlc(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)
@@ -43,12 +44,13 @@ case class HtlcId(channelId: ByteVector32, id: Long)
 /**
  * Local reputation for a given node.
  *
- * @param pending           Set of pending payments (payments may contain multiple HTLCs when using trampoline).
+ * @param pastScores        Scores from past HTLCs for each endorsement level.
+ * @param pending           Set of pending HTLCs.
  * @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.
+ * @param maxRelayDuration  Duration after which HTLCs are penalized for staying pending too long.
+ * @param pendingMultiplier How much to penalize pending HTLCs.
  */
-case class Reputation(pastScores: Array[PastScore], pending: mutable.Map[HtlcId, PendingPayment], halfLife: FiniteDuration, maxRelayDuration: FiniteDuration, pendingMultiplier: Double) {
+case class Reputation(pastScores: Array[PastScore], pending: mutable.Map[HtlcId, PendingHtlc], halfLife: FiniteDuration, maxRelayDuration: FiniteDuration, pendingMultiplier: Double) {
   private def decay(now: TimestampMilli, lastSettlementAt: TimestampMilli): Double = scala.math.pow(0.5, (now - lastSettlementAt) / halfLife)
 
   /**
@@ -88,10 +90,10 @@ case class Reputation(pastScores: Array[PastScore], pending: mutable.Map[HtlcId,
    * Register a pending relay.
    */
   def attempt(htlcId: HtlcId, fee: MilliSatoshi, endorsement: Int, now: TimestampMilli = TimestampMilli.now()): Unit =
-    pending(htlcId) = PendingPayment(fee, endorsement, now)
+    pending(htlcId) = PendingHtlc(fee, endorsement, now)
 
   /**
-   * When a payment is settled, we record whether it succeeded and how long it took.
+   * When a HTLC is settled, we record whether it succeeded and how long it took.
    */
   def record(htlcId: HtlcId, isSuccess: Boolean, now: TimestampMilli = TimestampMilli.now()): Unit =
     pending.remove(htlcId).foreach(p => {

eclair-core/src/main/scala/fr/acinq/eclair/reputation/ReputationRecorder.scala

@@ -17,7 +17,7 @@
 package fr.acinq.eclair.reputation
 
 import akka.actor.typed.eventstream.EventStream
-import akka.actor.typed.scaladsl.{ActorContext, Behaviors}
+import akka.actor.typed.scaladsl.Behaviors
 import akka.actor.typed.{ActorRef, Behavior}
 import fr.acinq.bitcoin.scalacompat.Crypto.PublicKey
 import fr.acinq.eclair.MilliSatoshi
@@ -37,21 +37,31 @@ object ReputationRecorder {
   private case class WrappedOutgoingHtlcFulfilled(fulfilled: OutgoingHtlcFulfilled) extends Command
   // @formatter:on
 
-  /** Confidence that the outgoing HTLC will succeed. */
+  /**
+   * @param confidence  Confidence that the outgoing HTLC will succeed (takes into account both upstream and downstream reputation).
+   * @param endorsement Endorsement level to set for the outgoing HTLC (takes into account upstream reputation only).
+   */
   case class Confidence(confidence: Double, endorsement: Int)
 
   def apply(config: Reputation.Config): Behavior[Command] =
     Behaviors.setup(context => {
       context.system.eventStream ! EventStream.Subscribe(context.messageAdapter(WrappedOutgoingHtlcAdded))
       context.system.eventStream ! EventStream.Subscribe(context.messageAdapter(WrappedOutgoingHtlcFailed))
       context.system.eventStream ! EventStream.Subscribe(context.messageAdapter(WrappedOutgoingHtlcFulfilled))
-      new ReputationRecorder(config, context).run()
+      new ReputationRecorder(config).run()
     })
 
-  case class PendingHtlc(add: UpdateAddHtlc, remoteNodeId: PublicKey, upstream: Upstream.Hot)
+  /**
+   * A pending outgoing HTLC.
+   *
+   * @param add        UpdateAddHtlc that contains an id for the HTLC and an endorsement value.
+   * @param upstream   The incoming node or nodes.
+   * @param downstream The outgoing node.
+   */
+  case class PendingHtlc(add: UpdateAddHtlc, upstream: Upstream.Hot, downstream: PublicKey)
 }
 
-class ReputationRecorder(config: Reputation.Config, context: ActorContext[ReputationRecorder.Command]) {
+class ReputationRecorder(config: Reputation.Config) {
   private val incomingReputations: mutable.Map[PublicKey, Reputation] = mutable.HashMap.empty
   private val outgoingReputations: mutable.Map[PublicKey, Reputation] = mutable.HashMap.empty
   private val pending: mutable.Map[HtlcId, PendingHtlc] = mutable.HashMap.empty
@@ -112,13 +122,17 @@ class ReputationRecorder(config: Reputation.Config, context: ActorContext[Reputa
           case channel: Hot.Channel =>
             incomingReputation(channel.receivedFrom).attempt(htlcId, fee, channel.add.endorsement)
           case trampoline: Hot.Trampoline =>
-            trampoline.received.foreach(channel =>
-              incomingReputation(channel.receivedFrom).attempt(htlcId, fee * channel.amountIn.toLong / trampoline.amountIn.toLong, channel.add.endorsement)
-            )
+            trampoline.received
+              .groupMapReduce(_.receivedFrom)(r => (r.add.amountMsat, r.add.endorsement)) {
+                case ((amount1, endorsement1), (amount2, endorsement2)) => (amount1 + amount2, endorsement1 min endorsement2)
+              }
+              .foreach { case (nodeId, (amount, endorsement)) =>
+                incomingReputation(nodeId).attempt(htlcId, fee * amount.toLong / trampoline.amountIn.toLong, endorsement)
+              }
           case _: Upstream.Local => ()
         }
         outgoingReputation(remoteNodeId).attempt(htlcId, fee, add.endorsement)
-        pending(htlcId) = PendingHtlc(add, remoteNodeId, upstream)
+        pending(htlcId) = PendingHtlc(add, upstream, remoteNodeId)
         Behaviors.same
 
       case WrappedOutgoingHtlcFailed(OutgoingHtlcFailed(fail)) =>
@@ -136,7 +150,7 @@ class ReputationRecorder(config: Reputation.Config, context: ActorContext[Reputa
               )
             case _: Upstream.Local => ()
           }
-          outgoingReputation(p.remoteNodeId).record(htlcId, isSuccess = false)
+          outgoingReputation(p.downstream).record(htlcId, isSuccess = false)
         })
         Behaviors.same
 
@@ -152,7 +166,7 @@ class ReputationRecorder(config: Reputation.Config, context: ActorContext[Reputa
               )
             case _: Upstream.Local => ()
           }
-          outgoingReputation(p.remoteNodeId).record(htlcId, isSuccess = true)
+          outgoingReputation(p.downstream).record(htlcId, isSuccess = true)
         })
         Behaviors.same
     }

eclair-core/src/main/scala/fr/acinq/eclair/reputation/package.scala

@@ -18,6 +18,6 @@ package fr.acinq.eclair
 
 package object reputation {
   implicit class EndorsementDouble(private val c: Double) extends AnyVal {
-    def toEndorsement: Int = (c * 7.999).toInt
+    def toEndorsement: Int = (c * Reputation.endorsementLevels).toInt
   }
 }

eclair-core/src/main/scala/fr/acinq/eclair/wire/protocol/LightningMessageTypes.scala

@@ -23,7 +23,6 @@ import fr.acinq.bitcoin.scalacompat.{BlockHash, ByteVector32, ByteVector64, OutP
 import fr.acinq.eclair.blockchain.fee.FeeratePerKw
 import fr.acinq.eclair.channel.{ChannelFlags, ChannelType}
 import fr.acinq.eclair.payment.relay.Relayer
-import fr.acinq.eclair.reputation.EndorsementDouble
 import fr.acinq.eclair.wire.protocol.ChannelReadyTlv.ShortChannelIdTlv
 import fr.acinq.eclair.{Alias, BlockHeight, CltvExpiry, CltvExpiryDelta, Feature, Features, InitFeature, MilliSatoshi, MilliSatoshiLong, RealShortChannelId, ShortChannelId, TimestampSecond, UInt64, isAsciiPrintable}
 import scodec.bits.ByteVector

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

@@ -180,7 +180,7 @@ object TestConstants {
           feeProportionalMillionths = 30),
         enforcementDelay = 10 minutes,
         asyncPaymentsParams = AsyncPaymentsParams(1008, CltvExpiryDelta(144)),
-        peerReputationConfig = Reputation.Config(enabled = false, 1 day, 10 seconds, 100),
+        peerReputationConfig = Reputation.Config(enabled = true, 1 day, 10 seconds, 100),
       ),
       db = TestDatabases.inMemoryDb(),
       autoReconnect = false,
@@ -372,7 +372,7 @@ object TestConstants {
           feeProportionalMillionths = 30),
         enforcementDelay = 10 minutes,
         asyncPaymentsParams = AsyncPaymentsParams(1008, CltvExpiryDelta(144)),
-        peerReputationConfig = Reputation.Config(enabled = false, 2 day, 20 seconds, 200),
+        peerReputationConfig = Reputation.Config(enabled = true, 2 day, 20 seconds, 200),
       ),
       db = TestDatabases.inMemoryDb(),
       autoReconnect = false,

eclair-core/src/test/scala/fr/acinq/eclair/payment/PaymentLifecycleSpec.scala

@@ -18,6 +18,7 @@ package fr.acinq.eclair.payment
 
 import akka.actor.ActorRef
 import akka.actor.FSM.{CurrentState, SubscribeTransitionCallBack, Transition}
+import akka.actor.typed.scaladsl.Behaviors
 import akka.actor.typed.scaladsl.adapter._
 import akka.testkit.{TestFSMRef, TestProbe}
 import fr.acinq.bitcoin.scalacompat.Crypto.PublicKey
@@ -38,7 +39,7 @@ import fr.acinq.eclair.payment.relay.Relayer.RelayFees
 import fr.acinq.eclair.payment.send.PaymentInitiator.SendPaymentConfig
 import fr.acinq.eclair.payment.send.PaymentLifecycle._
 import fr.acinq.eclair.payment.send.{ClearRecipient, PaymentLifecycle, Recipient}
-import fr.acinq.eclair.reputation.Reputation
+import fr.acinq.eclair.reputation.{Reputation, ReputationRecorder}
 import fr.acinq.eclair.router.Announcements.makeChannelUpdate
 import fr.acinq.eclair.router.BaseRouterSpec.{blindedRouteFromHops, channelAnnouncement, channelHopFromUpdate}
 import fr.acinq.eclair.router.Graph.PaymentWeightRatios
@@ -93,7 +94,13 @@ class PaymentLifecycleSpec extends BaseRouterSpec {
     val nodeParams = TestConstants.Alice.nodeParams.copy(nodeKeyManager = testNodeKeyManager, channelKeyManager = testChannelKeyManager)
     val cfg = SendPaymentConfig(id, parentId, Some(defaultExternalId), defaultPaymentHash, invoice.nodeId, Upstream.Local(id), Some(invoice), None, storeInDb, publishEvent, recordMetrics)
     val (routerForwarder, register, sender, monitor, eventListener, metricsListener) = (TestProbe(), TestProbe(), TestProbe(), TestProbe(), TestProbe(), TestProbe())
-    val paymentFSM = TestFSMRef(new PaymentLifecycle(nodeParams, cfg, routerForwarder.ref, register.ref, None))
+    val reputationRecorder = system.spawnAnonymous(Behaviors.receiveMessage[ReputationRecorder.GetConfidence](getConfidence => {
+      assert(getConfidence.upstream.isInstanceOf[Upstream.Local])
+      assert(getConfidence.downstream == b)
+      getConfidence.replyTo ! ReputationRecorder.Confidence(1.0, Reputation.maxEndorsement)
+      Behaviors.same
+    }))
+    val paymentFSM = TestFSMRef(new PaymentLifecycle(nodeParams, cfg, routerForwarder.ref, register.ref, Some(reputationRecorder)))
     paymentFSM ! SubscribeTransitionCallBack(monitor.ref)
     val CurrentState(_, WAITING_FOR_REQUEST) = monitor.expectMsgClass(classOf[CurrentState[_]])
     system.eventStream.subscribe(eventListener.ref, classOf[PaymentEvent])