Improve data stored in AuditDb

docs/release-notes/eclair-vnext.md

@@ -26,6 +26,19 @@ However, when using zero-conf, this event may be emitted before the `channel-con
 
 See #3237 for more details.
 
+### Major changes to the AuditDb
+
+We make a collection of backwards-incompatible changes to all tables of the `audit` database.
+The main change is that it is way more relevant to track statistics for peer nodes instead of individual channels, so we want to track the `node_id` associated with each event.
+We also track more data about transactions we make and relayed payments, to more easily score peers based on the fees we're earning vs the fees we're paying (for on-chain transactions or for liquidity purchases).
+
+Note that we cannot migrate existing data (since it is lacking information that we now need), so we simply rename older tables with a `_before_v14` suffix and create new ones.
+Past data will thus not be accessible through the APIs, but can be queried directly using SQL if necessary.
+It should be acceptable, since liquidity decisions should be taken based on relatively recent data (a few weeks) in order to be economically relevant (nodes that generated fees months ago but aren't generating any new fees since then are probably not good peers).
+
+We expose a now `relaystats` API that ranks peers based on the routing fees they're generating.
+See #3245 for more details.
+
 ### Channel jamming accountability
 
 We update our channel jamming mitigation to match the latest draft of the [spec](https://github.com/lightning/bolts/pull/1280).
@@ -47,6 +60,7 @@ eclair.relay.reserved-for-accountable = 0.0
 
 - `findroute`, `findroutetonode` and `findroutebetweennodes` now include a `maxCltvExpiryDelta` parameter (#3234)
 - `channel-opened` was removed from the websocket in favor of `channel-confirmed` and `channel-ready` (#3237)
+- `networkfees` and `channelstats` are removed in favor in `relaystats` (#3245)
 
 ### Miscellaneous improvements and bug fixes
 

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

@@ -35,7 +35,7 @@ import fr.acinq.eclair.blockchain.bitcoind.rpc.BitcoinCoreClient.{AddressType, D
 import fr.acinq.eclair.blockchain.fee.{ConfirmationTarget, FeeratePerByte, FeeratePerKw}
 import fr.acinq.eclair.channel._
 import fr.acinq.eclair.crypto.Sphinx
-import fr.acinq.eclair.db.AuditDb.{NetworkFee, Stats}
+import fr.acinq.eclair.db.AuditDb.RelayStats
 import fr.acinq.eclair.db.{IncomingPayment, OfferData, OutgoingPayment, OutgoingPaymentStatus}
 import fr.acinq.eclair.io.Peer.{GetPeerInfo, OpenChannelResponse, PeerInfo}
 import fr.acinq.eclair.io._
@@ -159,9 +159,9 @@ trait Eclair {
 
   def audit(from: TimestampSecond, to: TimestampSecond, paginated_opt: Option[Paginated])(implicit timeout: Timeout): Future[AuditResponse]
 
-  def networkFees(from: TimestampSecond, to: TimestampSecond)(implicit timeout: Timeout): Future[Seq[NetworkFee]]
+  def relayStats(remoteNodeId: PublicKey, from: TimestampSecond, to: TimestampSecond)(implicit timeout: Timeout): Future[RelayStats]
 
-  def channelStats(from: TimestampSecond, to: TimestampSecond, paginated_opt: Option[Paginated])(implicit timeout: Timeout): Future[Seq[Stats]]
+  def relayStats(from: TimestampSecond, to: TimestampSecond, paginated_opt: Option[Paginated])(implicit timeout: Timeout): Future[Seq[RelayStats]]
 
   def getInvoice(paymentHash: ByteVector32)(implicit timeout: Timeout): Future[Option[Invoice]]
 
@@ -596,12 +596,12 @@ class EclairImpl(val appKit: Kit) extends Eclair with Logging with SpendFromChan
     ))
   }
 
-  override def networkFees(from: TimestampSecond, to: TimestampSecond)(implicit timeout: Timeout): Future[Seq[NetworkFee]] = {
-    Future(appKit.nodeParams.db.audit.listNetworkFees(from.toTimestampMilli, to.toTimestampMilli))
+  override def relayStats(remoteNodeId: PublicKey, from: TimestampSecond, to: TimestampSecond)(implicit timeout: Timeout): Future[RelayStats] = {
+    Future(appKit.nodeParams.db.audit.relayStats(remoteNodeId, from.toTimestampMilli, to.toTimestampMilli))
   }
 
-  override def channelStats(from: TimestampSecond, to: TimestampSecond, paginated_opt: Option[Paginated])(implicit timeout: Timeout): Future[Seq[Stats]] = {
-    Future(appKit.nodeParams.db.audit.stats(from.toTimestampMilli, to.toTimestampMilli, paginated_opt))
+  override def relayStats(from: TimestampSecond, to: TimestampSecond, paginated_opt: Option[Paginated])(implicit timeout: Timeout): Future[Seq[RelayStats]] = {
+    Future(appKit.nodeParams.db.audit.relayStats(from.toTimestampMilli, to.toTimestampMilli, paginated_opt))
   }
 
   override def allInvoices(from: TimestampSecond, to: TimestampSecond, paginated_opt: Option[Paginated])(implicit timeout: Timeout): Future[Seq[Invoice]] = Future {

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

@@ -27,3 +27,10 @@ case class Paginated(count: Int, skip: Int) {
   require(count >= 0, "count must be a positive number")
   require(skip >= 0, "skip must be a positive number")
 }
+
+object Paginated {
+  def paginate[T](results: Seq[T], paginated_opt: Option[Paginated]): Seq[T] = paginated_opt match {
+    case Some(paginated) => results.slice(paginated.skip, paginated.skip + paginated.count)
+    case None => results
+  }
+}

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

@@ -21,8 +21,9 @@ 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.transactions.Transactions
 import fr.acinq.eclair.wire.protocol._
-import fr.acinq.eclair.{BlockHeight, CltvExpiry, Features, MilliSatoshi, RealShortChannelId, ShortChannelId}
+import fr.acinq.eclair.{BlockHeight, CltvExpiry, Features, MilliSatoshi, RealShortChannelId, ShortChannelId, TimestampMilli}
 
 /**
  * Created by PM on 17/08/2016.
@@ -92,10 +93,19 @@ case class ChannelLiquidityPurchased(channel: ActorRef, channelId: ByteVector32,
 
 case class ChannelErrorOccurred(channel: ActorRef, channelId: ByteVector32, remoteNodeId: PublicKey, error: ChannelError, isFatal: Boolean) extends ChannelEvent
 
-// NB: the fee should be set to 0 when we're not paying it.
-case class TransactionPublished(channelId: ByteVector32, remoteNodeId: PublicKey, tx: Transaction, miningFee: Satoshi, desc: String) extends ChannelEvent
+/**
+ * We published a transaction related to the given [[channelId]].
+ *
+ * @param localMiningFee        mining fee paid by us in the given [[tx]].
+ * @param remoteMiningFee       mining fee paid by our channel peer in the given [[tx]].
+ * @param liquidityPurchase_opt optional liquidity purchase included in this transaction.
+ */
+case class TransactionPublished(channelId: ByteVector32, remoteNodeId: PublicKey, tx: Transaction, localMiningFee: Satoshi, remoteMiningFee: Satoshi, desc: String, liquidityPurchase_opt: Option[LiquidityAds.PurchaseBasicInfo], timestamp: TimestampMilli = TimestampMilli.now()) extends ChannelEvent {
+  val miningFee: Satoshi = localMiningFee + remoteMiningFee
+  val feerate: FeeratePerKw = Transactions.fee2rate(miningFee, tx.weight())
+}
 
-case class TransactionConfirmed(channelId: ByteVector32, remoteNodeId: PublicKey, tx: Transaction) extends ChannelEvent
+case class TransactionConfirmed(channelId: ByteVector32, remoteNodeId: PublicKey, tx: Transaction, timestamp: TimestampMilli = TimestampMilli.now()) extends ChannelEvent
 
 // NB: this event is only sent when the channel is available.
 case class AvailableBalanceChanged(channel: ActorRef, channelId: ByteVector32, aliases: ShortIdAliases, commitments: Commitments, lastAnnouncement_opt: Option[ChannelAnnouncement]) extends ChannelEvent

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

@@ -971,11 +971,12 @@ object Helpers {
       }
     }
 
-    /** Compute the fee paid by a commitment transaction. */
-    def commitTxFee(commitInput: InputInfo, commitTx: Transaction, localPaysCommitTxFees: Boolean): Satoshi = {
+    /** Compute the fee paid by a commitment transaction. The first result is the fee paid by us, the second one is the fee paid by our peer. */
+    def commitTxFee(commitInput: InputInfo, commitTx: Transaction, localPaysCommitTxFees: Boolean): (Satoshi, Satoshi) = {
       require(commitTx.txIn.size == 1, "transaction must have only one input")
       require(commitTx.txIn.exists(txIn => txIn.outPoint == commitInput.outPoint), "transaction must spend the funding output")
-      if (localPaysCommitTxFees) commitInput.txOut.amount - commitTx.txOut.map(_.amount).sum else 0 sat
+      val commitFee = commitInput.txOut.amount - commitTx.txOut.map(_.amount).sum
+      if (localPaysCommitTxFees) (commitFee, 0 sat) else (0 sat, commitFee)
     }
 
     /** Return the confirmation target that should be used for our local commitment. */

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

@@ -51,7 +51,7 @@ trait DualFundingHandlers extends CommonFundingHandlers {
         // to publish and we may be able to RBF.
         wallet.publishTransaction(fundingTx.signedTx).onComplete {
           case Success(_) =>
-            context.system.eventStream.publish(TransactionPublished(dualFundedTx.fundingParams.channelId, remoteNodeId, fundingTx.signedTx, fundingTx.tx.localFees.truncateToSatoshi, "funding"))
+            context.system.eventStream.publish(TransactionPublished(dualFundedTx.fundingParams.channelId, remoteNodeId, fundingTx.signedTx, localMiningFee = fundingTx.tx.localFees.truncateToSatoshi, remoteMiningFee = fundingTx.tx.remoteFees.truncateToSatoshi, "funding", dualFundedTx.liquidityPurchase_opt))
             // We rely on Bitcoin Core ZMQ notifications to learn about transactions that appear in our mempool, but
             // it doesn't provide strong guarantees that we'll always receive an event. This can be an issue for 0-conf
             // funding transactions, where we end up delaying our channel_ready or splice_locked.

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

@@ -232,7 +232,8 @@ trait ErrorHandlers extends CommonHandlers {
 
   /** Publish 2nd-stage transactions for our local commitment. */
   def doPublish(lcp: LocalCommitPublished, txs: Closing.LocalClose.SecondStageTransactions, commitment: FullCommitment): Unit = {
-    val publishCommitTx = PublishFinalTx(lcp.commitTx, commitment.fundingInput, "commit-tx", Closing.commitTxFee(commitment.commitInput(channelKeys), lcp.commitTx, commitment.localChannelParams.paysCommitTxFees), None)
+    val (localCommitFee, _) = Closing.commitTxFee(commitment.commitInput(channelKeys), lcp.commitTx, commitment.localChannelParams.paysCommitTxFees)
+    val publishCommitTx = PublishFinalTx(lcp.commitTx, commitment.fundingInput, "commit-tx", localCommitFee, None)
     val publishAnchorTx_opt = txs.anchorTx_opt match {
       case Some(anchorTx) if !lcp.isConfirmed =>
         val confirmationTarget = Closing.confirmationTarget(commitment.localCommit, commitment.localCommitParams.dustLimit, commitment.commitmentFormat, nodeParams.onChainFeeConf)
@@ -274,7 +275,8 @@ trait ErrorHandlers extends CommonHandlers {
       case closing: DATA_CLOSING => nodeParams.onChainFeeConf.getClosingFeerate(nodeParams.currentBitcoinCoreFeerates, closing.maxClosingFeerate_opt)
       case _ => nodeParams.onChainFeeConf.getClosingFeerate(nodeParams.currentBitcoinCoreFeerates, maxClosingFeerateOverride_opt = None)
     }
-    context.system.eventStream.publish(TransactionPublished(d.channelId, remoteNodeId, commitTx, Closing.commitTxFee(commitments.commitInput(channelKeys), commitTx, d.commitments.localChannelParams.paysCommitTxFees), "remote-commit"))
+    val (localCommitFee, remoteCommitFee) = Closing.commitTxFee(commitments.commitInput(channelKeys), commitTx, d.commitments.localChannelParams.paysCommitTxFees)
+    context.system.eventStream.publish(TransactionPublished(d.channelId, remoteNodeId, commitTx, localCommitFee, remoteCommitFee, "remote-commit", None))
     val (remoteCommitPublished, closingTxs) = Closing.RemoteClose.claimCommitTxOutputs(channelKeys, commitments, commitments.remoteCommit, commitTx, closingFeerate, finalScriptPubKey, nodeParams.onChainFeeConf.spendAnchorWithoutHtlcs)
     val nextData = d match {
       case closing: DATA_CLOSING => closing.copy(remoteCommitPublished = Some(remoteCommitPublished))
@@ -296,7 +298,8 @@ trait ErrorHandlers extends CommonHandlers {
       case closing: DATA_CLOSING => nodeParams.onChainFeeConf.getClosingFeerate(nodeParams.currentBitcoinCoreFeerates, closing.maxClosingFeerate_opt)
       case _ => nodeParams.onChainFeeConf.getClosingFeerate(nodeParams.currentBitcoinCoreFeerates, maxClosingFeerateOverride_opt = None)
     }
-    context.system.eventStream.publish(TransactionPublished(d.channelId, remoteNodeId, commitTx, Closing.commitTxFee(commitment.commitInput(channelKeys), commitTx, d.commitments.localChannelParams.paysCommitTxFees), "next-remote-commit"))
+    val (localCommitFee, remoteCommitFee) = Closing.commitTxFee(commitment.commitInput(channelKeys), commitTx, d.commitments.localChannelParams.paysCommitTxFees)
+    context.system.eventStream.publish(TransactionPublished(d.channelId, remoteNodeId, commitTx, localCommitFee, remoteCommitFee, "next-remote-commit", None))
     val (remoteCommitPublished, closingTxs) = Closing.RemoteClose.claimCommitTxOutputs(channelKeys, commitment, remoteCommit, commitTx, closingFeerate, finalScriptPubKey, nodeParams.onChainFeeConf.spendAnchorWithoutHtlcs)
     val nextData = d match {
       case closing: DATA_CLOSING => closing.copy(nextRemoteCommitPublished = Some(remoteCommitPublished))
@@ -350,7 +353,8 @@ trait ErrorHandlers extends CommonHandlers {
         val dustLimit = commitment.localCommitParams.dustLimit
         val (revokedCommitPublished, closingTxs) = Closing.RevokedClose.claimCommitTxOutputs(d.commitments.channelParams, channelKeys, tx, commitmentNumber, remotePerCommitmentSecret, toSelfDelay, commitmentFormat, nodeParams.db.channels, dustLimit, nodeParams.currentBitcoinCoreFeerates, nodeParams.onChainFeeConf, finalScriptPubKey)
         log.warning("txid={} was a revoked commitment, publishing the penalty tx", tx.txid)
-        context.system.eventStream.publish(TransactionPublished(d.channelId, remoteNodeId, tx, Closing.commitTxFee(commitment.commitInput(channelKeys), tx, d.commitments.localChannelParams.paysCommitTxFees), "revoked-commit"))
+        val (localCommitFee, remoteCommitFee) = Closing.commitTxFee(commitment.commitInput(channelKeys), tx, d.commitments.localChannelParams.paysCommitTxFees)
+        context.system.eventStream.publish(TransactionPublished(d.channelId, remoteNodeId, tx, localCommitFee, remoteCommitFee, "revoked-commit", None))
         val exc = FundingTxSpent(d.channelId, tx.txid)
         val error = Error(d.channelId, exc.getMessage)
         val nextData = d match {
@@ -364,7 +368,8 @@ trait ErrorHandlers extends CommonHandlers {
       case None => d match {
         case d: DATA_WAIT_FOR_REMOTE_PUBLISH_FUTURE_COMMITMENT =>
           log.warning("they published a future commit (because we asked them to) in txid={}", tx.txid)
-          context.system.eventStream.publish(TransactionPublished(d.channelId, remoteNodeId, tx, Closing.commitTxFee(d.commitments.latest.commitInput(channelKeys), tx, d.commitments.localChannelParams.paysCommitTxFees), "future-remote-commit"))
+          val (localCommitFee, remoteCommitFee) = Closing.commitTxFee(d.commitments.latest.commitInput(channelKeys), tx, d.commitments.localChannelParams.paysCommitTxFees)
+          context.system.eventStream.publish(TransactionPublished(d.channelId, remoteNodeId, tx, localCommitFee, remoteCommitFee, "future-remote-commit", None))
           val remotePerCommitmentPoint = d.remoteChannelReestablish.myCurrentPerCommitmentPoint
           val commitKeys = d.commitments.latest.remoteKeys(channelKeys, remotePerCommitmentPoint)
           val closingFeerate = nodeParams.onChainFeeConf.getClosingFeerate(nodeParams.currentBitcoinCoreFeerates, maxClosingFeerateOverride_opt = None)

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

@@ -43,7 +43,7 @@ trait SingleFundingHandlers extends CommonFundingHandlers {
   def publishFundingTx(channelId: ByteVector32, fundingTx: Transaction, fundingTxFee: Satoshi, replyTo: akka.actor.typed.ActorRef[OpenChannelResponse]): Unit = {
     wallet.commit(fundingTx).onComplete {
       case Success(true) =>
-        context.system.eventStream.publish(TransactionPublished(channelId, remoteNodeId, fundingTx, fundingTxFee, "funding"))
+        context.system.eventStream.publish(TransactionPublished(channelId, remoteNodeId, fundingTx, localMiningFee = fundingTxFee, remoteMiningFee = 0 sat, "funding", None))
         replyTo ! OpenChannelResponse.Created(channelId, fundingTxId = fundingTx.txid, fundingTxFee)
       case Success(false) =>
         replyTo ! OpenChannelResponse.Rejected("couldn't publish funding tx")

eclair-core/src/main/scala/fr/acinq/eclair/channel/publish/MempoolTxMonitor.scala

@@ -19,7 +19,7 @@ package fr.acinq.eclair.channel.publish
 import akka.actor.typed.eventstream.EventStream
 import akka.actor.typed.scaladsl.{ActorContext, Behaviors, TimerScheduler}
 import akka.actor.typed.{ActorRef, Behavior}
-import fr.acinq.bitcoin.scalacompat.{ByteVector32, OutPoint, Satoshi, Transaction, TxId}
+import fr.acinq.bitcoin.scalacompat.{ByteVector32, OutPoint, Satoshi, SatoshiLong, Transaction, TxId}
 import fr.acinq.eclair.blockchain.CurrentBlockHeight
 import fr.acinq.eclair.blockchain.bitcoind.rpc.BitcoinCoreClient
 import fr.acinq.eclair.channel.publish.TxPublisher.{TxPublishContext, TxRejectedReason}
@@ -136,7 +136,7 @@ private class MempoolTxMonitor(nodeParams: NodeParams,
 
   private def waitForConfirmation(): Behavior[Command] = {
     context.system.eventStream ! EventStream.Subscribe(context.messageAdapter[CurrentBlockHeight](cbc => WrappedCurrentBlockHeight(cbc.blockHeight)))
-    context.system.eventStream ! EventStream.Publish(TransactionPublished(txPublishContext.channelId_opt.getOrElse(ByteVector32.Zeroes), txPublishContext.remoteNodeId, cmd.tx, cmd.fee, cmd.desc))
+    context.system.eventStream ! EventStream.Publish(TransactionPublished(txPublishContext.channelId_opt.getOrElse(ByteVector32.Zeroes), txPublishContext.remoteNodeId, cmd.tx, localMiningFee = cmd.fee, remoteMiningFee = 0 sat, cmd.desc, None))
     Behaviors.receiveMessagePartial {
       case WrappedCurrentBlockHeight(currentBlockHeight) =>
         timers.startSingleTimer(CheckTxConfirmationsKey, CheckTxConfirmations(currentBlockHeight), (1 + Random.nextLong(nodeParams.channelConf.maxTxPublishRetryDelay.toMillis)).millis)