Add event for failed payment relay (#3244)

We add an event when a payment could not be relayed and indicates that
we may need to add liquidity towards the next node. It is really hard
to figure it out in the context of a single payment though, so this
event does not by itself mean that liquidity should be allocated. The
listeners should collect several events and regularly query the state
of existing channels with our peers, and network graph data for remote
nodes, to create good heuristics for allocating liquidity. Otherwise,
it would be trivial for malicious senders to game routing nodes into
allocating liquidity "for free" towards them, which could result in
financial loss.

Author: t-bast

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

@@ -21,8 +21,8 @@ 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, HtlcFailureMessage, LiquidityAds, UpdateAddHtlc, UpdateFulfillHtlc}
-import fr.acinq.eclair.{BlockHeight, Features, MilliSatoshi, RealShortChannelId, ShortChannelId}
+import fr.acinq.eclair.wire.protocol._
+import fr.acinq.eclair.{BlockHeight, CltvExpiry, Features, MilliSatoshi, RealShortChannelId, ShortChannelId}
 
 /**
  * Created by PM on 17/08/2016.
@@ -106,9 +106,14 @@ case class LocalCommitConfirmed(channel: ActorRef, remoteNodeId: PublicKey, chan
 
 case class ChannelClosed(channel: ActorRef, channelId: ByteVector32, closingType: ClosingType, closingTxId: TxId, commitments: Commitments) extends ChannelEvent
 
-case class OutgoingHtlcAdded(add: UpdateAddHtlc, remoteNodeId: PublicKey, fee: MilliSatoshi)
+/** An outgoing HTLC was sent to our channel peer: we're waiting for it to be settled. */
+case class OutgoingHtlcAdded(add: UpdateAddHtlc, remoteNodeId: PublicKey, fee: MilliSatoshi) extends ChannelEvent
 
-sealed trait OutgoingHtlcSettled
+/** An outgoing HTLC could not be sent through the given channel. */
+case class OutgoingHtlcNotAdded(channelId: ByteVector32, remoteNodeId: PublicKey, paymentHash: ByteVector32, amount: MilliSatoshi, expiry: CltvExpiry, reason: ChannelException) extends ChannelEvent
+
+/** An outgoing HTLC was settled by our channel peer. */
+sealed trait OutgoingHtlcSettled extends ChannelEvent
 
 case class OutgoingHtlcFailed(fail: HtlcFailureMessage) extends OutgoingHtlcSettled
 

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

@@ -3243,6 +3243,7 @@ class Channel(val nodeParams: NodeParams, val channelKeys: ChannelKeys, val wall
     log.warning(s"${cause.getMessage} while processing cmd=${c.getClass.getSimpleName} in state=$stateName")
     val replyTo = if (c.replyTo == ActorRef.noSender) sender() else c.replyTo
     replyTo ! RES_ADD_FAILED(c, cause, channelUpdate)
+    context.system.eventStream.publish(OutgoingHtlcNotAdded(stateData.channelId, remoteNodeId, c.paymentHash, c.amount, c.cltvExpiry, cause))
     context.system.eventStream.publish(ChannelErrorOccurred(self, stateData.channelId, remoteNodeId, LocalError(cause), isFatal = false))
     stay()
   }

eclair-core/src/main/scala/fr/acinq/eclair/payment/PaymentEvents.scala

@@ -145,6 +145,19 @@ case class PaymentReceived(paymentHash: ByteVector32, parts: Seq[PaymentEvent.In
   val settledAt: TimestampMilli = parts.map(_.receivedAt).max // we use max here because we fulfill the payment only once we received all the parts
 }
 
+/**
+ * This event is emitted when we couldn't relay a payment and the reason is *likely* a liquidity issue.
+ * This can help figure out where liquidity is needed to earn more routing fees by funding channels.
+ *
+ * Note that this event is *not* emitted on *every* payment relay failure. This event does *not* guarantee that the
+ * failure was a liquidity issue, and malicious senders can force this event to be triggered for payments that they
+ * would not have fulfilled. Listeners must add their own heuristics and gather additional data in order to efficiently
+ * allocate liquidity and optimize their routing fees.
+ *
+ * @param fees fees we would have earned if we had successfully relayed that payment (can be gamed by malicious senders).
+ */
+case class PaymentNotRelayed(paymentHash: ByteVector32, remoteNodeId: PublicKey, fees: MilliSatoshi)
+
 case class PaymentMetadataReceived(paymentHash: ByteVector32, paymentMetadata: ByteVector)
 
 case class PaymentSettlingOnChain(id: UUID, channelId: ByteVector32, amount: MilliSatoshi, paymentHash: ByteVector32, timestamp: TimestampMilli = TimestampMilli.now())

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

@@ -30,7 +30,7 @@ import fr.acinq.eclair.io.Peer.ProposeOnTheFlyFundingResponse
 import fr.acinq.eclair.io.{Peer, PeerReadyNotifier}
 import fr.acinq.eclair.payment.Monitoring.{Metrics, Tags}
 import fr.acinq.eclair.payment.relay.Relayer.{OutgoingChannel, OutgoingChannelParams}
-import fr.acinq.eclair.payment.{ChannelPaymentRelayed, IncomingPaymentPacket, PaymentEvent}
+import fr.acinq.eclair.payment.{ChannelPaymentRelayed, IncomingPaymentPacket, PaymentEvent, PaymentNotRelayed}
 import fr.acinq.eclair.reputation.Reputation
 import fr.acinq.eclair.reputation.ReputationRecorder.GetConfidence
 import fr.acinq.eclair.wire.protocol.FailureMessageCodecs.createBadOnionFailure
@@ -210,6 +210,7 @@ class ChannelRelay private(nodeParams: NodeParams,
           case RelayFailure(cmdFail) =>
             Metrics.recordPaymentRelayFailed(Tags.FailureType(cmdFail), Tags.RelayType.Channel)
             context.log.info("rejecting htlc reason={}", cmdFail.reason)
+            channels.headOption.map(_._2.nextNodeId).foreach(nextNodeId => context.system.eventStream ! EventStream.Publish(PaymentNotRelayed(r.add.paymentHash, nextNodeId, fees = upstream.amountIn - r.amountToForward)))
             safeSendAndStop(r.add.channelId, cmdFail)
           case RelayNeedsFunding(nextNodeId, cmdFail) =>
             // Note that in the channel relay case, we don't have any outgoing onion shared secrets.
@@ -230,6 +231,7 @@ class ChannelRelay private(nodeParams: NodeParams,
         context.log.warn(s"couldn't resolve downstream channel $channelId, failing htlc #${upstream.add.id}")
         val cmdFail = makeCmdFailHtlc(upstream.add.id, UnknownNextPeer())
         Metrics.recordPaymentRelayFailed(Tags.FailureType(cmdFail), Tags.RelayType.Channel)
+        channels.headOption.map(_._2.nextNodeId).foreach(nextNodeId => context.system.eventStream ! EventStream.Publish(PaymentNotRelayed(r.add.paymentHash, nextNodeId, fees = upstream.amountIn - r.amountToForward)))
         safeSendAndStop(upstream.add.channelId, cmdFail)
 
       case WrappedAddResponse(addFailed: RES_ADD_FAILED[_]) =>

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

@@ -270,6 +270,9 @@ class NodeRelay private(nodeParams: NodeParams,
           rejectExtraHtlcPartialFunction orElse {
             case WrappedResolvedPaths(resolved) if resolved.isEmpty =>
               context.log.warn("rejecting trampoline payment to blinded paths: no usable blinded path")
+              payloadOut.outgoingBlindedPaths.map(_.route.firstNodeId).collectFirst { case n: EncodedNodeId.WithPublicKey => n.publicKey }.foreach(nodeId => {
+                context.system.eventStream ! EventStream.Publish(PaymentNotRelayed(paymentHash, nodeId, fees = upstream.amountIn - outgoingAmount(upstream, nextPayload)))
+              })
               rejectPayment(upstream, Some(UnknownNextPeer()))
               stopping()
             case WrappedResolvedPaths(resolved) =>
@@ -411,6 +414,7 @@ class NodeRelay private(nodeParams: NodeParams,
               context.log.info("trampoline payment failed, attempting on-the-fly funding")
               attemptOnTheFlyFunding(upstream, walletNodeId, recipient, nextPayload, e.failures, startedAt)
             case _ =>
+              context.system.eventStream ! EventStream.Publish(PaymentNotRelayed(paymentHash, recipient.nodeId, fees = upstream.amountIn - outgoingAmount(upstream, nextPayload)))
               rejectPayment(upstream, translateError(nodeParams, e.failures, upstream, nextPayload))
               recordRelayDuration(startedAt, isSuccess = false)
               stopping()

eclair-core/src/test/scala/fr/acinq/eclair/channel/states/e/NormalStateSpec.scala

@@ -210,12 +210,15 @@ class NormalStateSpec extends TestKitBaseClass with FixtureAnyFunSuiteLike with
 
   test("recv CMD_ADD_HTLC (insufficient funds)") { f =>
     import f._
+    val listener = TestProbe()
+    systemA.eventStream.subscribe(listener.ref, classOf[OutgoingHtlcNotAdded])
     val sender = TestProbe()
     val initialState = alice.stateData.asInstanceOf[DATA_NORMAL]
     val add = CMD_ADD_HTLC(sender.ref, initialState.commitments.availableBalanceForSend + 1.msat, randomBytes32(), CltvExpiryDelta(144).toCltvExpiry(currentBlockHeight), TestConstants.emptyOnionPacket, None, Reputation.Score.max(accountable = false), None, localOrigin(sender.ref))
     alice ! add
     val error = InsufficientFunds(channelId(alice), amount = add.amount, missing = 0 sat, reserve = 20000 sat, fees = 3900 sat)
     sender.expectMsg(RES_ADD_FAILED(add, error, Some(initialState.channelUpdate)))
+    assert(listener.expectMsgType[OutgoingHtlcNotAdded].reason == error)
     alice2bob.expectNoMessage(100 millis)
   }
 
@@ -333,6 +336,8 @@ class NormalStateSpec extends TestKitBaseClass with FixtureAnyFunSuiteLike with
 
   test("recv CMD_ADD_HTLC (over remote max inflight htlc value)", Tag(ChannelStateTestsTags.AliceLowMaxHtlcValueInFlight)) { f =>
     import f._
+    val listener = TestProbe()
+    systemB.eventStream.subscribe(listener.ref, classOf[OutgoingHtlcNotAdded])
     val sender = TestProbe()
     val initialState = bob.stateData.asInstanceOf[DATA_NORMAL]
     assert(initialState.commitments.latest.localCommitParams.maxHtlcValueInFlight == UInt64(initialState.commitments.latest.capacity.toMilliSatoshi.toLong))
@@ -341,6 +346,7 @@ class NormalStateSpec extends TestKitBaseClass with FixtureAnyFunSuiteLike with
     bob ! add
     val error = HtlcValueTooHighInFlight(channelId(bob), maximum = UInt64(150_000_000), actual = 151_000_000 msat)
     sender.expectMsg(RES_ADD_FAILED(add, error, Some(initialState.channelUpdate)))
+    assert(listener.expectMsgType[OutgoingHtlcNotAdded].reason == error)
     bob2alice.expectNoMessage(100 millis)
   }
 

eclair-core/src/test/scala/fr/acinq/eclair/payment/relay/ChannelRelayerSpec.scala

@@ -34,7 +34,7 @@ import fr.acinq.eclair.crypto.Sphinx
 import fr.acinq.eclair.io.{Peer, PeerReadyManager, Switchboard}
 import fr.acinq.eclair.payment.IncomingPaymentPacket.ChannelRelayPacket
 import fr.acinq.eclair.payment.relay.ChannelRelayer._
-import fr.acinq.eclair.payment.{ChannelPaymentRelayed, IncomingPaymentPacket, PaymentPacketSpec}
+import fr.acinq.eclair.payment.{ChannelPaymentRelayed, IncomingPaymentPacket, PaymentNotRelayed, PaymentPacketSpec}
 import fr.acinq.eclair.reputation.{Reputation, ReputationRecorder}
 import fr.acinq.eclair.router.Announcements
 import fr.acinq.eclair.wire.protocol.BlindedRouteData.PaymentRelayData
@@ -539,6 +539,9 @@ class ChannelRelayerSpec extends ScalaTestWithActorTestKit(ConfigFactory.load("a
   test("fail to relay when there is a local error") { f =>
     import f._
 
+    val eventListener = TestProbe[PaymentNotRelayed]()
+    system.eventStream ! EventStream.Subscribe(eventListener.ref)
+
     val channelId1 = channelIds(realScid1)
     val payload = ChannelRelay.Standard(realScid1, outgoingAmount, outgoingExpiry, upgradeAccountability = false)
     val r = createValidIncomingPacket(payload)
@@ -564,6 +567,7 @@ class ChannelRelayerSpec extends ScalaTestWithActorTestKit(ConfigFactory.load("a
       val fwd = expectFwdAdd(register, channelIds(realScid1), outgoingAmount, outgoingExpiry, outAccountable = false)
       fwd.message.replyTo ! RES_ADD_FAILED(fwd.message, testCase.exc, Some(testCase.update))
       expectFwdFail(register, r.add.channelId, CMD_FAIL_HTLC(r.add.id, FailureReason.LocalFailure(testCase.failure), None, commit = true))
+      assert(eventListener.expectMessageType[PaymentNotRelayed].remoteNodeId == outgoingNodeId)
     }
   }
 

eclair-core/src/test/scala/fr/acinq/eclair/payment/relay/NodeRelayerSpec.scala

@@ -473,6 +473,9 @@ class NodeRelayerSpec extends ScalaTestWithActorTestKit(ConfigFactory.load("appl
   test("fail to relay because outgoing balance isn't sufficient (low fees)") { f =>
     import f._
 
+    val listener = TestProbe[PaymentNotRelayed]()
+    system.eventStream ! EventStream.Subscribe(listener.ref)
+
     // Receive an upstream multi-part payment.
     val (nodeRelayer, _) = f.createNodeRelay(incomingMultiPart.head)
     incomingMultiPart.foreach(p => nodeRelayer ! NodeRelay.Relay(p, randomKey().publicKey, 0.01))
@@ -492,8 +495,8 @@ class NodeRelayerSpec extends ScalaTestWithActorTestKit(ConfigFactory.load("appl
       assert(fwd.message == CMD_FAIL_HTLC(p.add.id, FailureReason.LocalFailure(TrampolineFeeInsufficient()), Some(FailureAttributionData(p.receivedAt, Some(incomingMultiPart.last.receivedAt))), commit = true))
     }
 
+    assert(listener.expectMessageType[PaymentNotRelayed].remoteNodeId == getPeerInfo.nodeId)
     register.expectNoMessage(100 millis)
-    eventListener.expectNoMessage(100 millis)
   }
 
   test("fail to relay because outgoing balance isn't sufficient (high fees)") { f =>