Add event for failed payment relay (#3244)

diff --git a/eclair-core/src/main/scala/fr/acinq/eclair/channel/ChannelEvents.scala b/eclair-core/src/main/scala/fr/acinq/eclair/channel/ChannelEvents.scala
index aa719f5..890b93e 100644
--- a/eclair-core/src/main/scala/fr/acinq/eclair/channel/ChannelEvents.scala
+++ b/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
 
diff --git a/eclair-core/src/main/scala/fr/acinq/eclair/channel/fsm/Channel.scala b/eclair-core/src/main/scala/fr/acinq/eclair/channel/fsm/Channel.scala
index 699cd5a..28ae5b4 100644
--- a/eclair-core/src/main/scala/fr/acinq/eclair/channel/fsm/Channel.scala
+++ b/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()
   }
diff --git a/eclair-core/src/main/scala/fr/acinq/eclair/payment/PaymentEvents.scala b/eclair-core/src/main/scala/fr/acinq/eclair/payment/PaymentEvents.scala
index 2315cf3..7874753 100644
--- a/eclair-core/src/main/scala/fr/acinq/eclair/payment/PaymentEvents.scala
+++ b/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())
diff --git a/eclair-core/src/main/scala/fr/acinq/eclair/payment/relay/ChannelRelay.scala b/eclair-core/src/main/scala/fr/acinq/eclair/payment/relay/ChannelRelay.scala
index 130e4b7..97f9877 100644
--- a/eclair-core/src/main/scala/fr/acinq/eclair/payment/relay/ChannelRelay.scala
+++ b/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[_]) =>
diff --git a/eclair-core/src/main/scala/fr/acinq/eclair/payment/relay/NodeRelay.scala b/eclair-core/src/main/scala/fr/acinq/eclair/payment/relay/NodeRelay.scala
index 3f0da94..e2b873c 100644
--- a/eclair-core/src/main/scala/fr/acinq/eclair/payment/relay/NodeRelay.scala
+++ b/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()
diff --git a/eclair-core/src/test/scala/fr/acinq/eclair/channel/states/e/NormalStateSpec.scala b/eclair-core/src/test/scala/fr/acinq/eclair/channel/states/e/NormalStateSpec.scala
index d0e8027..e965e02 100644
--- a/eclair-core/src/test/scala/fr/acinq/eclair/channel/states/e/NormalStateSpec.scala
+++ b/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)
   }
 
diff --git a/eclair-core/src/test/scala/fr/acinq/eclair/payment/relay/ChannelRelayerSpec.scala b/eclair-core/src/test/scala/fr/acinq/eclair/payment/relay/ChannelRelayerSpec.scala
index 8120339..8f93d78 100644
--- a/eclair-core/src/test/scala/fr/acinq/eclair/payment/relay/ChannelRelayerSpec.scala
+++ b/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)
     }
   }
 
diff --git a/eclair-core/src/test/scala/fr/acinq/eclair/payment/relay/NodeRelayerSpec.scala b/eclair-core/src/test/scala/fr/acinq/eclair/payment/relay/NodeRelayerSpec.scala
index 4368b0f..2eef4c1 100644
--- a/eclair-core/src/test/scala/fr/acinq/eclair/payment/relay/NodeRelayerSpec.scala
+++ b/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 =>