Add local reputation

docs/release-notes/eclair-vnext.md

@@ -24,6 +24,25 @@ Existing `static_remote_key` channels will continue to work. You can override th
 
 Eclair will not allow remote peers to open new obsolete channels that do not support `option_static_remotekey`.
 
+### Local reputation and HTLC endorsement
+
+To protect against jamming attacks, eclair gives a reputation to its neighbors and uses to decide if a HTLC should be relayed given how congested is the outgoing channel.
+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.
+
+To configure, edit `eclair.conf`:
+```eclair.conf
+eclair.local-reputation {
+    # Reputation decays with the following half life to emphasize recent behavior.
+    half-life = 7 days
+    # HTLCs that stay pending for longer than this get penalized
+    good-htlc-duration = 12 seconds
+    # How much to penalize pending HLTCs. A pending HTLC is considered equivalent to this many fast-failing HTLCs.
+    pending-multiplier = 1000
+}
+```
+
 ### API changes
 
 <insert changes>

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

@@ -547,6 +547,15 @@ eclair {
     enabled = true // enable automatic purges of expired invoices from the database
     interval = 24 hours // interval between expired invoice purges
   }
+
+  local-reputation {
+    # Reputation decays with the following half life to emphasize recent behavior.
+    half-life = 7 days
+    # HTLCs that stay pending for longer than this get penalized
+    good-htlc-duration = 12 seconds # 95% of successful payments settle in less than 12 seconds, only the slowest 5% will be penalized.
+    # How much to penalize pending HLTCs. A pending HTLC is considered equivalent to this many fast-failing HTLCs.
+    pending-multiplier = 1000
+  }
 }
 
 akka {

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

@@ -31,6 +31,7 @@ import fr.acinq.eclair.io.MessageRelay.{RelayAll, RelayChannelsOnly, RelayPolicy
 import fr.acinq.eclair.io.PeerConnection
 import fr.acinq.eclair.message.OnionMessages.OnionMessageConfig
 import fr.acinq.eclair.payment.relay.Relayer.{AsyncPaymentsParams, RelayFees, RelayParams}
+import fr.acinq.eclair.reputation.Reputation.ReputationConfig
 import fr.acinq.eclair.router.Announcements.AddressException
 import fr.acinq.eclair.router.Graph.{HeuristicsConstants, WeightRatios}
 import fr.acinq.eclair.router.Router._
@@ -87,7 +88,8 @@ case class NodeParams(nodeKeyManager: NodeKeyManager,
                       blockchainWatchdogSources: Seq[String],
                       onionMessageConfig: OnionMessageConfig,
                       purgeInvoicesInterval: Option[FiniteDuration],
-                      revokedHtlcInfoCleanerConfig: RevokedHtlcInfoCleaner.Config) {
+                      revokedHtlcInfoCleanerConfig: RevokedHtlcInfoCleaner.Config,
+                      localReputationConfig: ReputationConfig) {
   val privateKey: Crypto.PrivateKey = nodeKeyManager.nodeKey.privateKey
 
   val nodeId: PublicKey = nodeKeyManager.nodeId
@@ -611,7 +613,12 @@ object NodeParams extends Logging {
       revokedHtlcInfoCleanerConfig = RevokedHtlcInfoCleaner.Config(
         batchSize = config.getInt("db.revoked-htlc-info-cleaner.batch-size"),
         interval = FiniteDuration(config.getDuration("db.revoked-htlc-info-cleaner.interval").getSeconds, TimeUnit.SECONDS)
-      )
+      ),
+      localReputationConfig = ReputationConfig(
+        FiniteDuration(config.getDuration("local-reputation.half-life").getSeconds, TimeUnit.SECONDS),
+        FiniteDuration(config.getDuration("local-reputation.good-htlc-duration").getSeconds, TimeUnit.SECONDS),
+        config.getDouble("local-reputation.pending-multiplier"),
+      ),
     )
   }
 }

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

@@ -44,6 +44,7 @@ import fr.acinq.eclair.payment.offer.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
@@ -360,7 +361,8 @@ class Setup(val datadir: File,
       offerManager = system.spawn(Behaviors.supervise(OfferManager(nodeParams, router, paymentTimeout = 1 minute)).onFailure(typed.SupervisorStrategy.resume), name = "offer-manager")
       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")
-      relayer = system.actorOf(SimpleSupervisor.props(Relayer.props(nodeParams, router, register, paymentHandler, triggerer, Some(postRestartCleanUpInitialized)), "relayer", SupervisorStrategy.Resume))
+      reputationRecorder = system.spawn(Behaviors.supervise(ReputationRecorder(nodeParams.localReputationConfig, Map.empty)).onFailure(typed.SupervisorStrategy.resume), name = "reputation-recorder")
+      relayer = system.actorOf(SimpleSupervisor.props(Relayer.props(nodeParams, router, register, paymentHandler, triggerer, reputationRecorder, 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/ChannelExceptions.scala

@@ -118,6 +118,8 @@ case class ExpiryTooBig                            (override val channelId: Byte
 case class HtlcValueTooSmall                       (override val channelId: ByteVector32, minimum: MilliSatoshi, actual: MilliSatoshi) extends ChannelException(channelId, s"htlc value too small: minimum=$minimum actual=$actual")
 case class HtlcValueTooHighInFlight                (override val channelId: ByteVector32, maximum: MilliSatoshi, actual: MilliSatoshi) extends ChannelException(channelId, s"in-flight htlcs hold too much value: maximum=$maximum actual=$actual")
 case class TooManyAcceptedHtlcs                    (override val channelId: ByteVector32, maximum: Long) extends ChannelException(channelId, s"too many accepted htlcs: maximum=$maximum")
+case class TooManySmallHtlcs                       (override val channelId: ByteVector32, number: Long, below: MilliSatoshi) extends ChannelException(channelId, s"too many small htlcs: $number HTLCs below $below")
+case class ConfidenceTooLow                        (override val channelId: ByteVector32, confidence: Double, occupancy: Double) extends ChannelException(channelId, s"confidence too low: confidence=$confidence occupancy=$occupancy")
 case class LocalDustHtlcExposureTooHigh            (override val channelId: ByteVector32, maximum: Satoshi, actual: MilliSatoshi) extends ChannelException(channelId, s"dust htlcs hold too much value: maximum=$maximum actual=$actual")
 case class RemoteDustHtlcExposureTooHigh           (override val channelId: ByteVector32, maximum: Satoshi, actual: MilliSatoshi) extends ChannelException(channelId, s"dust htlcs hold too much value: maximum=$maximum actual=$actual")
 case class InsufficientFunds                       (override val channelId: ByteVector32, amount: MilliSatoshi, missing: Satoshi, reserve: Satoshi, fees: Satoshi) extends ChannelException(channelId, s"insufficient funds: missing=$missing reserve=$reserve fees=$fees")

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

@@ -429,7 +429,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
@@ -488,7 +488,8 @@ case class Commitment(fundingTxIndex: Long,
     if (allowedHtlcValueInFlight < htlcValueInFlight) {
       return Left(HtlcValueTooHighInFlight(params.channelId, maximum = allowedHtlcValueInFlight, actual = htlcValueInFlight))
     }
-    if (Seq(params.localParams.maxAcceptedHtlcs, params.remoteParams.maxAcceptedHtlcs).min < outgoingHtlcs.size) {
+    val maxAcceptedHtlcs = params.localParams.maxAcceptedHtlcs.min(params.remoteParams.maxAcceptedHtlcs)
+    if (maxAcceptedHtlcs < outgoingHtlcs.size) {
       return Left(TooManyAcceptedHtlcs(params.channelId, maximum = Seq(params.localParams.maxAcceptedHtlcs, params.remoteParams.maxAcceptedHtlcs).min))
     }
 
@@ -505,6 +506,18 @@ case class Commitment(fundingTxIndex: Long,
       return Left(RemoteDustHtlcExposureTooHigh(params.channelId, maxDustExposure, remoteDustExposureAfterAdd))
     }
 
+    // Jamming protection
+    // Must be the last checks so that they can be ignored for shadow deployment.
+    for ((amountMsat, i) <- outgoingHtlcs.toSeq.map(_.amountMsat).sorted.zipWithIndex) {
+      if ((amountMsat.toLong < 1) || (math.log(amountMsat.toLong.toDouble) * maxAcceptedHtlcs / math.log(params.localParams.maxHtlcValueInFlightMsat.toLong.toDouble / maxAcceptedHtlcs) < i)) {
+        return Left(TooManySmallHtlcs(params.channelId, number = i + 1, below = amountMsat))
+      }
+    }
+    val occupancy = (outgoingHtlcs.size.toDouble / maxAcceptedHtlcs).max(htlcValueInFlight.toLong.toDouble / allowedHtlcValueInFlight.toLong.toDouble)
+    if (confidence + 0.05 < occupancy) {
+      return Left(ConfidenceTooLow(params.channelId, confidence, occupancy))
+    }
+
     Right(())
   }
 
@@ -552,6 +565,14 @@ case class Commitment(fundingTxIndex: Long,
       return Left(TooManyAcceptedHtlcs(params.channelId, maximum = params.localParams.maxAcceptedHtlcs))
     }
 
+    // Jamming protection
+    // Must be the last checks so that they can be ignored for shadow deployment.
+    for ((amountMsat, i) <- incomingHtlcs.toSeq.map(_.amountMsat).sorted.zipWithIndex) {
+      if ((amountMsat.toLong < 1) || (math.log(amountMsat.toLong.toDouble) * params.localParams.maxAcceptedHtlcs / math.log(params.localParams.maxHtlcValueInFlightMsat.toLong.toDouble / params.localParams.maxAcceptedHtlcs) < i)) {
+        return Left(TooManySmallHtlcs(params.channelId, number = i + 1, below = amountMsat))
+      }
+    }
+
     Right(())
   }
 
@@ -835,7 +856,7 @@ case class Commitments(params: 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)
@@ -859,12 +880,28 @@ case class Commitments(params: ChannelParams,
     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, params, changes1, feerates, feeConf))
-      .collectFirst { case Left(f) => Left(f) }
+    val canSendAdds = active.map(_.canSendAdd(add.amountMsat, params, changes1, feerates, feeConf, cmd.confidence))
+    // Log only for jamming protection.
+    canSendAdds.collectFirst {
+      case Left(f: TooManySmallHtlcs) =>
+        log.info("TooManySmallHtlcs: {} outgoing HTLCs are below {}}", f.number, f.below)
+        Metrics.dropHtlc(f, Tags.Directions.Outgoing)
+      case Left(f: ConfidenceTooLow) =>
+        log.info("ConfidenceTooLow: confidence is {}% while channel is {}% full", (100 * f.confidence).toInt, (100 * f.occupancy).toInt)
+        Metrics.dropHtlc(f, Tags.Directions.Outgoing)
+    }
+    canSendAdds.flatMap { // TODO: We ignore jamming protection, delete this flatMap to activate jamming protection.
+        case Left(_: TooManySmallHtlcs) | Left(_: ConfidenceTooLow) => None
+        case x => Some(x)
+      }
+      .collectFirst { case Left(f) =>
+        Metrics.dropHtlc(f, Tags.Directions.Outgoing)
+        Left(f)
+      }
       .getOrElse(Right(copy(changes = changes1, originChannels = originChannels1), add))
   }
 
-  def receiveAdd(add: UpdateAddHtlc, feerates: FeeratesPerKw, feeConf: OnChainFeeConf): Either[ChannelException, Commitments] = {
+  def receiveAdd(add: UpdateAddHtlc, feerates: FeeratesPerKw, feeConf: OnChainFeeConf)(implicit log: LoggingAdapter): Either[ChannelException, Commitments] = {
     if (add.id != changes.remoteNextHtlcId) {
       return Left(UnexpectedHtlcId(channelId, expected = changes.remoteNextHtlcId, actual = add.id))
     }
@@ -877,8 +914,21 @@ case class Commitments(params: ChannelParams,
 
     val changes1 = changes.addRemoteProposal(add).copy(remoteNextHtlcId = changes.remoteNextHtlcId + 1)
     // we verify that this htlc is allowed in every active commitment
-    active.map(_.canReceiveAdd(add.amountMsat, params, changes1, feerates, feeConf))
-      .collectFirst { case Left(f) => Left(f) }
+    val canReceiveAdds = active.map(_.canReceiveAdd(add.amountMsat, params, changes1, feerates, feeConf))
+    // Log only for jamming protection.
+    canReceiveAdds.collectFirst {
+      case Left(f: TooManySmallHtlcs) =>
+        log.info("TooManySmallHtlcs: {} incoming HTLCs are below {}}", f.number, f.below)
+        Metrics.dropHtlc(f, Tags.Directions.Incoming)
+    }
+    canReceiveAdds.flatMap { // TODO: We ignore jamming protection, delete this flatMap to activate jamming protection.
+        case Left(_: TooManySmallHtlcs) | Left(_: ConfidenceTooLow) => None
+        case x => Some(x)
+      }
+      .collectFirst { case Left(f) =>
+        Metrics.dropHtlc(f, Tags.Directions.Incoming)
+        Left(f)
+      }
       .getOrElse(Right(copy(changes = changes1)))
   }
 

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"