Endorse htlc and local reputation

docs/release-notes/eclair-vnext.md

@@ -25,6 +25,33 @@ Attribution data also provides hold times from payment relayers, both for fulfil
 Support is enabled by default.
 It can be disabled by setting `eclair.features.option_attribution_data = disabled`.
 
+### Local reputation and HTLC endorsement
+
+To protect against jamming attacks, eclair gives a reputation to its neighbors and uses it to decide if a HTLC should be relayed given how congested the outgoing channel is.
+The reputation is basically how much this node paid us in fees divided by how much they should have paid us for the liquidity and slots that they blocked.
+The reputation is per incoming node and endorsement level.
+The confidence that the HTLC will be fulfilled is transmitted to the next node using the endorsement TLV of the `update_add_htlc` message.
+Note that HTLCs that are considered dangerous are still relayed: this is the first phase of a network-wide experimentation aimed at collecting data.
+
+To configure, edit `eclair.conf`:
+
+```eclair.conf
+// We assign reputations to our peers to prioritize payments during congestion.
+// The reputation is computed as fees paid divided by what should have been paid if all payments were successful.
+eclair.relay.peer-reputation {
+    // Set this parameter to false to disable the reputation algorithm and simply relay the incoming endorsement
+    // 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 = 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 = 200 // A pending payment counts as two hundred failed ones.
+}
+```
+
 ### API changes
 
 - `listoffers` now returns more details about each offer.

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

@@ -251,6 +251,23 @@ eclair {
       // Number of blocks before the incoming HTLC expires that an async payment must be triggered by the receiver
       cancel-safety-before-timeout-blocks = 144
     }
+
+    // We assign reputation to our peers to prioritize payments during congestion.
+    // The reputation is computed as fees paid divided by what should have been paid if all payments were successful.
+    peer-reputation {
+      // Set this parameter to false to disable the reputation algorithm and simply relay the incoming endorsement
+      // 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 = 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. We want it to be as close as possible to the true cost of a worst case
+      // HTLC (max-cltv-delta / max-relay-duration, around 100000 with default parameters) while still being comparable
+      // to the number of HTLCs received per peer during twice the half life.
+      pending-multiplier = 200 // A pending payment counts as two hundred failed ones.
+    }
   }
 
   on-chain-fees {

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

@@ -33,6 +33,7 @@ import fr.acinq.eclair.message.OnionMessages.OnionMessageConfig
 import fr.acinq.eclair.payment.offer.OffersConfig
 import fr.acinq.eclair.payment.relay.OnTheFlyFunding
 import fr.acinq.eclair.payment.relay.Relayer.{AsyncPaymentsParams, RelayFees, RelayParams}
+import fr.acinq.eclair.reputation.Reputation
 import fr.acinq.eclair.router.Announcements.AddressException
 import fr.acinq.eclair.router.Graph.{HeuristicsConstants, PaymentWeightRatios}
 import fr.acinq.eclair.router.Router._
@@ -641,7 +642,13 @@ object NodeParams extends Logging {
         privateChannelFees = getRelayFees(config.getConfig("relay.fees.private-channels")),
         minTrampolineFees = getRelayFees(config.getConfig("relay.fees.min-trampoline")),
         enforcementDelay = FiniteDuration(config.getDuration("relay.fees.enforcement-delay").getSeconds, TimeUnit.SECONDS),
-        asyncPaymentsParams = AsyncPaymentsParams(asyncPaymentHoldTimeoutBlocks, asyncPaymentCancelSafetyBeforeTimeoutBlocks)
+        asyncPaymentsParams = AsyncPaymentsParams(asyncPaymentHoldTimeoutBlocks, asyncPaymentCancelSafetyBeforeTimeoutBlocks),
+        peerReputationConfig = Reputation.Config(
+          enabled = config.getBoolean("relay.peer-reputation.enabled"),
+          halfLife = FiniteDuration(config.getDuration("relay.peer-reputation.half-life").getSeconds, TimeUnit.SECONDS),
+          maxRelayDuration = FiniteDuration(config.getDuration("relay.peer-reputation.max-relay-duration").getSeconds, TimeUnit.SECONDS),
+          pendingMultiplier = config.getDouble("relay.peer-reputation.pending-multiplier"),
+        ),
       ),
       db = database,
       autoReconnect = config.getBoolean("auto-reconnect"),

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

@@ -45,6 +45,7 @@ import fr.acinq.eclair.payment.offer.{DefaultOfferHandler, OfferManager}
 import fr.acinq.eclair.payment.receive.PaymentHandler
 import fr.acinq.eclair.payment.relay.{AsyncPaymentTriggerer, PostRestartHtlcCleaner, Relayer}
 import fr.acinq.eclair.payment.send.{Autoprobe, PaymentInitiator}
+import fr.acinq.eclair.reputation.ReputationRecorder
 import fr.acinq.eclair.router._
 import fr.acinq.eclair.tor.{Controller, TorProtocolHandler}
 import fr.acinq.eclair.wire.protocol.NodeAddress
@@ -379,7 +380,12 @@ class Setup(val datadir: File,
       paymentHandler = system.actorOf(SimpleSupervisor.props(PaymentHandler.props(nodeParams, register, offerManager), "payment-handler", SupervisorStrategy.Resume))
       triggerer = system.spawn(Behaviors.supervise(AsyncPaymentTriggerer()).onFailure(typed.SupervisorStrategy.resume), name = "async-payment-triggerer")
       peerReadyManager = system.spawn(Behaviors.supervise(PeerReadyManager()).onFailure(typed.SupervisorStrategy.restart), name = "peer-ready-manager")
-      relayer = system.actorOf(SimpleSupervisor.props(Relayer.props(nodeParams, router, register, paymentHandler, Some(postRestartCleanUpInitialized)), "relayer", SupervisorStrategy.Resume))
+      reputationRecorder_opt = if (nodeParams.relayParams.peerReputationConfig.enabled) {
+        Some(system.spawn(Behaviors.supervise(ReputationRecorder(nodeParams.relayParams.peerReputationConfig)).onFailure(typed.SupervisorStrategy.resume), name = "reputation-recorder"))
+      } else {
+        None
+      }
+      relayer = system.actorOf(SimpleSupervisor.props(Relayer.props(nodeParams, router, register, paymentHandler, reputationRecorder_opt, Some(postRestartCleanUpInitialized)), "relayer", SupervisorStrategy.Resume))
       _ = relayer ! PostRestartHtlcCleaner.Init(channels)
       // Before initializing the switchboard (which re-connects us to the network) and the user-facing parts of the system,
       // we want to make sure the handler for post-restart broken HTLCs has finished initializing.

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

@@ -24,6 +24,7 @@ import fr.acinq.eclair.channel.LocalFundingStatus.DualFundedUnconfirmedFundingTx
 import fr.acinq.eclair.channel.fund.InteractiveTxBuilder._
 import fr.acinq.eclair.channel.fund.{InteractiveTxBuilder, InteractiveTxSigningSession}
 import fr.acinq.eclair.io.Peer
+import fr.acinq.eclair.reputation.Reputation
 import fr.acinq.eclair.transactions.CommitmentSpec
 import fr.acinq.eclair.transactions.Transactions._
 import fr.acinq.eclair.wire.protocol.{ChannelAnnouncement, ChannelReady, ChannelReestablish, ChannelUpdate, ClosingSigned, CommitSig, FailureReason, FundingCreated, FundingSigned, Init, LiquidityAds, OnionRoutingPacket, OpenChannel, OpenDualFundedChannel, Shutdown, SpliceInit, Stfu, TxInitRbf, TxSignatures, UpdateAddHtlc, UpdateFailHtlc, UpdateFailMalformedHtlc, UpdateFulfillHtlc}
@@ -153,13 +154,17 @@ object Upstream {
     case class Channel(add: UpdateAddHtlc, receivedAt: TimestampMilli, receivedFrom: PublicKey) extends Hot {
       override val amountIn: MilliSatoshi = add.amountMsat
       val expiryIn: CltvExpiry = add.cltvExpiry
+
+      override def toString: String = s"Channel(amountIn=$amountIn, 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 toString: String = s"Trampoline(${received.map(_.toString).mkString(",")})"
     }
   }
 
@@ -216,7 +221,7 @@ final case class CMD_ADD_HTLC(replyTo: ActorRef,
                               cltvExpiry: CltvExpiry,
                               onion: OnionRoutingPacket,
                               nextPathKey_opt: Option[PublicKey],
-                              confidence: Double,
+                              reputationScore: Reputation.Score,
                               fundingFee_opt: Option[LiquidityAds.FundingFee],
                               origin: Origin.Hot,
                               commit: Boolean = false) extends HasReplyToCommand with ForbiddenCommandDuringQuiescenceNegotiation with ForbiddenCommandWhenQuiescent

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

@@ -21,7 +21,7 @@ import fr.acinq.bitcoin.scalacompat.Crypto.PublicKey
 import fr.acinq.bitcoin.scalacompat.{ByteVector32, Satoshi, Transaction, TxId}
 import fr.acinq.eclair.blockchain.fee.FeeratePerKw
 import fr.acinq.eclair.channel.Helpers.Closing.ClosingType
-import fr.acinq.eclair.wire.protocol.{ChannelAnnouncement, ChannelUpdate, LiquidityAds}
+import fr.acinq.eclair.wire.protocol.{ChannelAnnouncement, ChannelUpdate, HtlcFailureMessage, LiquidityAds, UpdateAddHtlc, UpdateFulfillHtlc}
 import fr.acinq.eclair.{BlockHeight, Features, MilliSatoshi, RealShortChannelId, ShortChannelId}
 
 /**
@@ -104,3 +104,11 @@ case class ChannelPersisted(channel: ActorRef, remoteNodeId: PublicKey, channelI
 case class LocalCommitConfirmed(channel: ActorRef, remoteNodeId: PublicKey, channelId: ByteVector32, refundAtBlock: BlockHeight) extends ChannelEvent
 
 case class ChannelClosed(channel: ActorRef, channelId: ByteVector32, closingType: ClosingType, commitments: Commitments) extends ChannelEvent
+
+case class OutgoingHtlcAdded(add: UpdateAddHtlc, remoteNodeId: PublicKey, upstream: Upstream.Hot, fee: MilliSatoshi)
+
+sealed trait OutgoingHtlcSettled
+
+case class OutgoingHtlcFailed(fail: HtlcFailureMessage) extends OutgoingHtlcSettled
+
+case class OutgoingHtlcFulfilled(fulfill: UpdateFulfillHtlc) extends OutgoingHtlcSettled

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

@@ -34,6 +34,7 @@ case class RemoteError(e: protocol.Error) extends ChannelError
 // @formatter:on
 
 class ChannelException(val channelId: ByteVector32, message: String) extends RuntimeException(message)
+class ChannelJammingException(override val channelId: ByteVector32, message: String) extends ChannelException(channelId, message)
 
 // @formatter:off
 case class InvalidChainHash                        (override val channelId: ByteVector32, local: BlockHash, remote: BlockHash) extends ChannelException(channelId, s"invalid chainHash (local=$local remote=$remote)")
@@ -150,4 +151,6 @@ case class CommandUnavailableInThisState           (override val channelId: Byte
 case class ForbiddenDuringSplice                   (override val channelId: ByteVector32, command: String) extends ChannelException(channelId, s"cannot process $command while splicing")
 case class ForbiddenDuringQuiescence               (override val channelId: ByteVector32, command: String) extends ChannelException(channelId, s"cannot process $command while quiescent")
 case class ConcurrentRemoteSplice                  (override val channelId: ByteVector32) extends ChannelException(channelId, "splice attempt canceled, remote initiated splice before us")
+case class TooManySmallHtlcs                       (override val channelId: ByteVector32, number: Long, below: MilliSatoshi) extends ChannelJammingException(channelId, s"too many small htlcs: $number HTLCs below $below")
+case class ConfidenceTooLow                        (override val channelId: ByteVector32, confidence: Double, occupancy: Double) extends ChannelJammingException(channelId, s"confidence too low: confidence=$confidence occupancy=$occupancy")
 // @formatter:on

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

@@ -11,6 +11,7 @@ import fr.acinq.eclair.channel.fsm.Channel.ChannelConf
 import fr.acinq.eclair.crypto.ShaChain
 import fr.acinq.eclair.crypto.keymanager.{ChannelKeys, LocalCommitmentKeys, RemoteCommitmentKeys}
 import fr.acinq.eclair.payment.OutgoingPaymentPacket
+import fr.acinq.eclair.reputation.Reputation
 import fr.acinq.eclair.router.Announcements
 import fr.acinq.eclair.transactions.Transactions._
 import fr.acinq.eclair.transactions._
@@ -452,7 +453,7 @@ case class Commitment(fundingTxIndex: Long,
     localCommit.spec.htlcs.collect(DirectedHtlc.incoming).filter(nearlyExpired)
   }
 
-  def canSendAdd(amount: MilliSatoshi, params: ChannelParams, changes: CommitmentChanges, feerates: FeeratesPerKw, feeConf: OnChainFeeConf): Either[ChannelException, Unit] = {
+  def canSendAdd(amount: MilliSatoshi, params: ChannelParams, changes: CommitmentChanges, feerates: FeeratesPerKw, feeConf: OnChainFeeConf, confidence: Double): Either[ChannelException, Unit] = {
     // we allowed mismatches between our feerates and our remote's as long as commitments didn't contain any HTLC at risk
     // we need to verify that we're not disagreeing on feerates anymore before offering new HTLCs
     // NB: there may be a pending update_fee that hasn't been applied yet that needs to be taken into account
@@ -528,7 +529,9 @@ case class Commitment(fundingTxIndex: Long,
       return Left(RemoteDustHtlcExposureTooHigh(params.channelId, maxDustExposure, remoteDustExposureAfterAdd))
     }
 
-    Right(())
+    // Jamming protection
+    // Must be the last checks so that they can be ignored for shadow deployment.
+    Reputation.checkChannelOccupancy(outgoingHtlcs.toSeq, params, confidence)
   }
 
   def canReceiveAdd(amount: MilliSatoshi, params: ChannelParams, changes: CommitmentChanges, feerates: FeeratesPerKw, feeConf: OnChainFeeConf): Either[ChannelException, Unit] = {
@@ -880,7 +883,7 @@ case class Commitments(channelParams: ChannelParams,
    * @param cmd add HTLC command
    * @return either Left(failure, error message) where failure is a failure message (see BOLT #4 and the Failure Message class) or Right(new commitments, updateAddHtlc)
    */
-  def sendAdd(cmd: CMD_ADD_HTLC, currentHeight: BlockHeight, channelConf: ChannelConf, feerates: FeeratesPerKw, feeConf: OnChainFeeConf): Either[ChannelException, (Commitments, UpdateAddHtlc)] = {
+  def sendAdd(cmd: CMD_ADD_HTLC, currentHeight: BlockHeight, channelConf: ChannelConf, feerates: FeeratesPerKw, feeConf: OnChainFeeConf)(implicit log: LoggingAdapter): Either[ChannelException, (Commitments, UpdateAddHtlc)] = {
     // we must ensure we're not relaying htlcs that are already expired, otherwise the downstream channel will instantly close
     // NB: we add a 3 blocks safety to reduce the probability of running into this when our bitcoin node is slightly outdated
     val minExpiry = CltvExpiry(currentHeight + 3)
@@ -899,14 +902,31 @@ case class Commitments(channelParams: ChannelParams,
       return Left(HtlcValueTooSmall(channelId, minimum = htlcMinimum, actual = cmd.amount))
     }
 
-    val add = UpdateAddHtlc(channelId, changes.localNextHtlcId, cmd.amount, cmd.paymentHash, cmd.cltvExpiry, cmd.onion, cmd.nextPathKey_opt, cmd.confidence, cmd.fundingFee_opt)
+    val add = UpdateAddHtlc(channelId, changes.localNextHtlcId, cmd.amount, cmd.paymentHash, cmd.cltvExpiry, cmd.onion, cmd.nextPathKey_opt, cmd.reputationScore.endorsement, cmd.fundingFee_opt)
     // we increment the local htlc index and add an entry to the origins map
     val changes1 = changes.addLocalProposal(add).copy(localNextHtlcId = changes.localNextHtlcId + 1)
     val originChannels1 = originChannels + (add.id -> cmd.origin)
     // we verify that this htlc is allowed in every active commitment
-    active.map(_.canSendAdd(add.amountMsat, channelParams, changes1, feerates, feeConf))
-      .collectFirst { case Left(f) => Left(f) }
-      .getOrElse(Right(copy(changes = changes1, originChannels = originChannels1), add))
+    val failures = active.map(_.canSendAdd(add.amountMsat, channelParams, changes1, feerates, feeConf, cmd.reputationScore.incomingConfidence)).collect { case Left(f) => f }
+    if (failures.isEmpty) {
+      Right(copy(changes = changes1, originChannels = originChannels1), add)
+    } else if (failures.forall(_.isInstanceOf[ChannelJammingException])) {
+      // We ignore jamming protection for now, but we log which HTLCs would be dropped if it was enabled.
+      val failure = failures.collectFirst { case f: ChannelJammingException => f }.get
+      Metrics.dropHtlc(failure, Tags.Directions.Outgoing)
+      failure match {
+        case f: TooManySmallHtlcs => log.info("TooManySmallHtlcs: {} outgoing HTLCs are below {}", f.number, f.below)
+        case f: ConfidenceTooLow => log.info("ConfidenceTooLow: confidence is {}% while channel is {}% full", (100 * f.confidence).toInt, (100 * f.occupancy).toInt)
+        case _ => ()
+      }
+      Right(copy(changes = changes1, originChannels = originChannels1), add)
+    } else {
+      // In most cases, the same failure will be returned for every commitment. Even if that's not the case, we can only
+      // send a single failure message to our peer, so we use the one that applies to the most recent active commitment.
+      val failure = failures.filterNot(_.isInstanceOf[ChannelJammingException]).head
+      Metrics.dropHtlc(failure, Tags.Directions.Outgoing)
+      Left(failure)
+    }
   }
 
   def receiveAdd(add: UpdateAddHtlc, feerates: FeeratesPerKw, feeConf: OnChainFeeConf): Either[ChannelException, Commitments] = {

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

@@ -35,6 +35,7 @@ object Monitoring {
     val RemoteFeeratePerByte = Kamon.histogram("channels.remote-feerate-per-byte")
     val Splices = Kamon.histogram("channels.splices", "Splices")
     val ProcessMessage = Kamon.timer("channels.messages-processed")
+    val HtlcDropped = Kamon.counter("channels.htlc-dropped")
 
     def recordHtlcsInFlight(remoteSpec: CommitmentSpec, previousRemoteSpec: CommitmentSpec): Unit = {
       for (direction <- Tags.Directions.Incoming :: Tags.Directions.Outgoing :: Nil) {
@@ -75,6 +76,10 @@ object Monitoring {
         Metrics.Splices.withTag(Tags.Origin, Tags.Origins.Remote).withTag(Tags.SpliceType, Tags.SpliceTypes.SpliceCpfp).record(Math.abs(fundingParams.remoteContribution.toLong))
       }
     }
+
+    def dropHtlc(reason: ChannelException, direction: String): Unit = {
+      HtlcDropped.withTag(Tags.Reason, reason.getClass.getSimpleName).withTag(Tags.Direction, direction).increment()
+    }
   }
 
   object Tags {
@@ -85,6 +90,7 @@ object Monitoring {
     val State = "state"
     val CommitmentFormat = "commitment-format"
     val SpliceType = "splice-type"
+    val Reason = "reason"
 
     object Events {
       val Created = "created"