Allow disabling the reputation recorder

We must allow node operators to run plain blip-0004 without our
experimental reputation recorder. It's also a stop-gap in case
there is a bug in it that can be exploited.

We also refactor to introduce more types and documentation, without
changing the reputation algorithm itself.

We also fix a few issues mentioned on the main PR comments.

Author: t-bast

docs/release-notes/eclair-vnext.md

@@ -26,20 +26,28 @@ Eclair will not allow remote peers to open new obsolete channels that do not sup
 
 ### 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. 
+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
-eclair.local-reputation {
-    # Reputation decays with the following half life to emphasize recent behavior.
+// 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.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 = 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
+    // 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.
 }
 ```
 

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

@@ -238,16 +238,19 @@ eclair {
       cancel-safety-before-timeout-blocks = 144
     }
 
-    // We assign reputations to our peers to prioritize HTLCs during congestion.
-    // The reputation is computed as fees paid divided by what should have been paid if all HTLCs were successful.
+    // 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 = 7 days
-      // HTLCs that stay pending for longer than this get penalized
-      max-htlc-relay-duration = 12 seconds
-      // Pending HTLCs 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 HTLCs counts as a thousand failed ones.
+      // 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.
     }
   }
 

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

@@ -31,7 +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.reputation.Reputation
 import fr.acinq.eclair.router.Announcements.AddressException
 import fr.acinq.eclair.router.Graph.{HeuristicsConstants, WeightRatios}
 import fr.acinq.eclair.router.Router._
@@ -563,10 +563,11 @@ object NodeParams extends Logging {
         minTrampolineFees = getRelayFees(config.getConfig("relay.fees.min-trampoline")),
         enforcementDelay = FiniteDuration(config.getDuration("relay.fees.enforcement-delay").getSeconds, TimeUnit.SECONDS),
         asyncPaymentsParams = AsyncPaymentsParams(asyncPaymentHoldTimeoutBlocks, asyncPaymentCancelSafetyBeforeTimeoutBlocks),
-        peerReputationConfig = ReputationConfig(
-          FiniteDuration(config.getDuration("relay.peer-reputation.half-life").getSeconds, TimeUnit.SECONDS),
-          FiniteDuration(config.getDuration("relay.peer-reputation.max-htlc-relay-duration").getSeconds, TimeUnit.SECONDS),
-          config.getDouble("relay.peer-reputation.pending-multiplier"),
+        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,

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

@@ -361,8 +361,12 @@ 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")
-      reputationRecorder = system.spawn(Behaviors.supervise(ReputationRecorder(nodeParams.relayParams.peerReputationConfig, 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))
+      reputationRecorder_opt = if (nodeParams.relayParams.peerReputationConfig.enabled) {
+        Some(system.spawn(Behaviors.supervise(ReputationRecorder(nodeParams.relayParams.peerReputationConfig, Map.empty)).onFailure(typed.SupervisorStrategy.resume), name = "reputation-recorder"))
+      } else {
+        None
+      }
+      relayer = system.actorOf(SimpleSupervisor.props(Relayer.props(nodeParams, router, register, paymentHandler, triggerer, 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/payment/relay/ChannelRelay.scala

@@ -45,6 +45,7 @@ object ChannelRelay {
 
   // @formatter:off
   sealed trait Command
+  private case object DoRelay extends Command
   private case class WrappedConfidence(confidence: Double) extends Command
   private case class WrappedForwardFailure(failure: Register.ForwardFailure[CMD_ADD_HTLC]) extends Command
   private case class WrappedAddResponse(res: CommandResponse[CMD_ADD_HTLC]) extends Command
@@ -58,9 +59,9 @@ object ChannelRelay {
 
   def apply(nodeParams: NodeParams,
             register: ActorRef,
-            reputationRecorder: typed.ActorRef[ReputationRecorder.StandardCommand],
+            reputationRecorder_opt: Option[typed.ActorRef[ReputationRecorder.ChannelRelayCommand]],
             channels: Map[ByteVector32, Relayer.OutgoingChannel],
-            originNode:PublicKey,
+            originNode: PublicKey,
             relayId: UUID,
             r: IncomingPaymentPacket.ChannelRelayPacket): Behavior[Command] =
     Behaviors.setup { context =>
@@ -70,10 +71,17 @@ object ChannelRelay {
         paymentHash_opt = Some(r.add.paymentHash),
         nodeAlias_opt = Some(nodeParams.alias))) {
         val upstream = Upstream.Hot.Channel(r.add.removeUnknownTlvs(), TimestampMilli.now(), originNode)
-        reputationRecorder ! GetConfidence(context.messageAdapter[ReputationRecorder.Confidence](confidence => WrappedConfidence(confidence.value)), originNode, r.add.endorsement, relayId, r.relayFeeMsat)
+        reputationRecorder_opt match {
+          case Some(reputationRecorder) =>
+            reputationRecorder ! GetConfidence(context.messageAdapter[ReputationRecorder.Confidence](confidence => WrappedConfidence(confidence.value)), originNode, r.add.endorsement, relayId, r.relayFeeMsat)
+          case None =>
+            val confidence = (r.add.endorsement + 0.5) / 8
+            context.self ! WrappedConfidence(confidence)
+        }
         Behaviors.receiveMessagePartial {
           case WrappedConfidence(confidence) =>
-            new ChannelRelay(nodeParams, register, reputationRecorder, channels, r, upstream, confidence, context, relayId).relay(Seq.empty)
+            context.self ! DoRelay
+            new ChannelRelay(nodeParams, register, reputationRecorder_opt, channels, r, upstream, confidence, context, relayId).relay(Seq.empty)
         }
       }
     }
@@ -115,7 +123,7 @@ object ChannelRelay {
  */
 class ChannelRelay private(nodeParams: NodeParams,
                            register: ActorRef,
-                           reputationRecorder: typed.ActorRef[ReputationRecorder.StandardCommand],
+                           reputationRecorder_opt: Option[typed.ActorRef[ReputationRecorder.ChannelRelayCommand]],
                            channels: Map[ByteVector32, Relayer.OutgoingChannel],
                            r: IncomingPaymentPacket.ChannelRelayPacket,
                            upstream: Upstream.Hot.Channel,
@@ -131,6 +139,8 @@ class ChannelRelay private(nodeParams: NodeParams,
   private case class PreviouslyTried(channelId: ByteVector32, failure: RES_ADD_FAILED[ChannelException])
 
   def relay(previousFailures: Seq[PreviouslyTried]): Behavior[Command] = {
+    Behaviors.receiveMessagePartial {
+      case DoRelay =>
         if (previousFailures.isEmpty) {
           context.log.info("relaying htlc #{} from channelId={} to requestedShortChannelId={} nextNode={}", r.add.id, r.add.channelId, r.payload.outgoingChannelId, nextNodeId_opt.getOrElse(""))
         }
@@ -139,13 +149,14 @@ class ChannelRelay private(nodeParams: NodeParams,
           case RelayFailure(cmdFail) =>
             Metrics.recordPaymentRelayFailed(Tags.FailureType(cmdFail), Tags.RelayType.Channel)
             context.log.info("rejecting htlc reason={}", cmdFail.reason)
-            reputationRecorder ! CancelRelay(upstream.receivedFrom, r.add.endorsement, relayId)
+            reputationRecorder_opt.foreach(_ ! CancelRelay(upstream.receivedFrom, r.add.endorsement, relayId))
             safeSendAndStop(r.add.channelId, cmdFail)
           case RelaySuccess(selectedChannelId, cmdAdd) =>
             context.log.info("forwarding htlc #{} from channelId={} to channelId={}", r.add.id, r.add.channelId, selectedChannelId)
             register ! Register.Forward(forwardFailureAdapter, selectedChannelId, cmdAdd)
             waitForAddResponse(selectedChannelId, previousFailures)
         }
+    }
   }
 
   def waitForAddResponse(selectedChannelId: ByteVector32, previousFailures: Seq[PreviouslyTried]): Behavior[Command] =
@@ -154,11 +165,12 @@ class ChannelRelay private(nodeParams: NodeParams,
         context.log.warn(s"couldn't resolve downstream channel $channelId, failing htlc #${upstream.add.id}")
         val cmdFail = CMD_FAIL_HTLC(upstream.add.id, Right(UnknownNextPeer()), commit = true)
         Metrics.recordPaymentRelayFailed(Tags.FailureType(cmdFail), Tags.RelayType.Channel)
-        reputationRecorder ! CancelRelay(upstream.receivedFrom, r.add.endorsement, relayId)
+        reputationRecorder_opt.foreach(_ ! CancelRelay(upstream.receivedFrom, r.add.endorsement, relayId))
         safeSendAndStop(upstream.add.channelId, cmdFail)
 
       case WrappedAddResponse(addFailed: RES_ADD_FAILED[_]) =>
         context.log.info("attempt failed with reason={}", addFailed.t.getClass.getSimpleName)
+        context.self ! DoRelay
         relay(previousFailures :+ PreviouslyTried(selectedChannelId, addFailed))
 
       case WrappedAddResponse(r: RES_SUCCESS[_]) =>
@@ -331,7 +343,7 @@ class ChannelRelay private(nodeParams: NodeParams,
   }
 
   private def recordRelayDuration(isSuccess: Boolean): Unit = {
-    reputationRecorder ! RecordResult(upstream.receivedFrom, r.add.endorsement, relayId, isSuccess)
+    reputationRecorder_opt.foreach(_ ! RecordResult(upstream.receivedFrom, r.add.endorsement, relayId, isSuccess))
     Metrics.RelayedPaymentDuration
       .withTag(Tags.Relay, Tags.RelayType.Channel)
       .withTag(Tags.Success, isSuccess)

eclair-core/src/main/scala/fr/acinq/eclair/payment/relay/ChannelRelayer.scala

@@ -59,7 +59,7 @@ object ChannelRelayer {
 
   def apply(nodeParams: NodeParams,
             register: ActorRef,
-            reputationRecorder: typed.ActorRef[ReputationRecorder.StandardCommand],
+            reputationRecorder_opt: Option[typed.ActorRef[ReputationRecorder.ChannelRelayCommand]],
             channels: Map[ByteVector32, Relayer.OutgoingChannel] = Map.empty,
             scid2channels: Map[ShortChannelId, ByteVector32] = Map.empty,
             node2channels: mutable.MultiDict[PublicKey, ByteVector32] = mutable.MultiDict.empty): Behavior[Command] =
@@ -81,7 +81,7 @@ object ChannelRelayer {
               case None => Map.empty
             }
             context.log.debug(s"spawning a new handler with relayId=$relayId to nextNodeId={} with channels={}", nextNodeId_opt.getOrElse(""), nextChannels.keys.mkString(","))
-            context.spawn(ChannelRelay.apply(nodeParams, register, reputationRecorder, nextChannels, originNode, relayId, channelRelayPacket), name = relayId.toString)
+            context.spawn(ChannelRelay.apply(nodeParams, register, reputationRecorder_opt, nextChannels, originNode, relayId, channelRelayPacket), name = relayId.toString)
             Behaviors.same
 
           case GetOutgoingChannels(replyTo, Relayer.GetOutgoingChannels(enabledOnly)) =>
@@ -102,14 +102,14 @@ object ChannelRelayer {
             context.log.debug("adding mappings={} to channelId={}", mappings.keys.mkString(","), channelId)
             val scid2channels1 = scid2channels ++ mappings
             val node2channels1 = node2channels.addOne(remoteNodeId, channelId)
-            apply(nodeParams, register, reputationRecorder, channels1, scid2channels1, node2channels1)
+            apply(nodeParams, register, reputationRecorder_opt, channels1, scid2channels1, node2channels1)
 
           case WrappedLocalChannelDown(LocalChannelDown(_, channelId, shortIds, remoteNodeId)) =>
             context.log.debug(s"removed local channel info for channelId=$channelId localAlias=${shortIds.localAlias}")
             val channels1 = channels - channelId
             val scid2Channels1 = scid2channels - shortIds.localAlias -- shortIds.real.toOption
             val node2channels1 = node2channels.subtractOne(remoteNodeId, channelId)
-            apply(nodeParams, register, reputationRecorder, channels1, scid2Channels1, node2channels1)
+            apply(nodeParams, register, reputationRecorder_opt, channels1, scid2Channels1, node2channels1)
 
           case WrappedAvailableBalanceChanged(AvailableBalanceChanged(_, channelId, shortIds, commitments)) =>
             val channels1 = channels.get(channelId) match {
@@ -118,7 +118,7 @@ object ChannelRelayer {
                 channels + (channelId -> c.copy(commitments = commitments))
               case None => channels // we only consider the balance if we have the channel_update
             }
-            apply(nodeParams, register, reputationRecorder, channels1, scid2channels, node2channels)
+            apply(nodeParams, register, reputationRecorder_opt, channels1, scid2channels, node2channels)
 
         }
       }