Skip to content

Improve channel and payment events #3246

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
t-bast wants to merge 1 commit into
base: master
Choose a base branch
from
Draft

Improve channel and payment events #3246

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 14 additions & 4 deletions eclair-core/src/main/scala/fr/acinq/eclair/channel/ChannelEvents.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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
Expand Down
7 changes: 4 additions & 3 deletions eclair-core/src/main/scala/fr/acinq/eclair/channel/Helpers.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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. */
Expand Down
2 changes: 1 addition & 1 deletion eclair-core/src/main/scala/fr/acinq/eclair/channel/fsm/DualFundingHandlers.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
15 changes: 10 additions & 5 deletions eclair-core/src/main/scala/fr/acinq/eclair/channel/fsm/ErrorHandlers.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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))
Expand All @@ -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))
Expand Down Expand Up @@ -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 {
Expand All @@ -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)
Expand Down
2 changes: 1 addition & 1 deletion eclair-core/src/main/scala/fr/acinq/eclair/channel/fsm/SingleFundingHandlers.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down
4 changes: 2 additions & 2 deletions eclair-core/src/main/scala/fr/acinq/eclair/channel/publish/MempoolTxMonitor.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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}
Expand Down Expand Up @@ -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)
Expand Down
26 changes: 13 additions & 13 deletions eclair-core/src/main/scala/fr/acinq/eclair/db/DbEventHandler.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,7 @@ import fr.acinq.eclair.channel._
import fr.acinq.eclair.db.DbEventHandler.ChannelEvent
import fr.acinq.eclair.payment.Monitoring.{Metrics => PaymentMetrics, Tags => PaymentTags}
import fr.acinq.eclair.payment._
import fr.acinq.eclair.{Logs, NodeParams}
import fr.acinq.eclair.{Logs, NodeParams, TimestampMilli}

/**
* This actor sits at the interface between our event stream and the database.
Expand Down Expand Up @@ -90,8 +90,8 @@ class DbEventHandler(nodeParams: NodeParams) extends Actor with DiagnosticActorL
incoming.foreach(p => channelsDb.updateChannelMeta(p.channelId, ChannelEvent.EventType.PaymentReceived))
outgoing.foreach(p => channelsDb.updateChannelMeta(p.channelId, ChannelEvent.EventType.PaymentSent))
case ChannelPaymentRelayed(_, incoming, outgoing) =>
channelsDb.updateChannelMeta(incoming.channelId, ChannelEvent.EventType.PaymentReceived)
channelsDb.updateChannelMeta(outgoing.channelId, ChannelEvent.EventType.PaymentSent)
incoming.foreach(i => channelsDb.updateChannelMeta(i.channelId, ChannelEvent.EventType.PaymentReceived))
outgoing.foreach(o => channelsDb.updateChannelMeta(o.channelId, ChannelEvent.EventType.PaymentSent))
case OnTheFlyFundingPaymentRelayed(_, incoming, outgoing) =>
incoming.foreach(p => channelsDb.updateChannelMeta(p.channelId, ChannelEvent.EventType.PaymentReceived))
outgoing.foreach(p => channelsDb.updateChannelMeta(p.channelId, ChannelEvent.EventType.PaymentSent))
Expand Down Expand Up @@ -124,7 +124,7 @@ class DbEventHandler(nodeParams: NodeParams) extends Actor with DiagnosticActorL
case ChannelStateChanged(_, channelId, _, remoteNodeId, WAIT_FOR_CHANNEL_READY | WAIT_FOR_DUAL_FUNDING_READY, NORMAL, Some(commitments)) =>
ChannelMetrics.ChannelLifecycleEvents.withTag(ChannelTags.Event, ChannelTags.Events.Created).increment()
val event = ChannelEvent.EventType.Created
auditDb.add(ChannelEvent(channelId, remoteNodeId, commitments.latest.fundingTxId, commitments.latest.capacity, commitments.localChannelParams.isChannelOpener, !commitments.announceChannel, event))
auditDb.add(ChannelEvent(channelId, remoteNodeId, commitments.latest.fundingTxId, commitments.latest.commitmentFormat.toString, commitments.latest.capacity, commitments.localChannelParams.isChannelOpener, !commitments.announceChannel, event.label))
channelsDb.updateChannelMeta(channelId, event)
case ChannelStateChanged(_, channelId, _, _, OFFLINE, SYNCING, _) =>
channelsDb.updateChannelMeta(channelId, ChannelEvent.EventType.Connected)
Expand All @@ -141,7 +141,7 @@ class DbEventHandler(nodeParams: NodeParams) extends Actor with DiagnosticActorL
case 0 => ChannelEvent.EventType.Confirmed
case _ => ChannelEvent.EventType.Spliced
}
auditDb.add(ChannelEvent(e.channelId, e.remoteNodeId, e.fundingTxId, e.commitments.latest.capacity, e.commitments.localChannelParams.isChannelOpener, !e.commitments.announceChannel, event))
auditDb.add(ChannelEvent(e.channelId, e.remoteNodeId, e.fundingTxId, e.commitments.latest.commitmentFormat.toString, e.commitments.latest.capacity, e.commitments.localChannelParams.isChannelOpener, !e.commitments.announceChannel, event.label))

case e: ChannelClosed =>
ChannelMetrics.ChannelLifecycleEvents.withTag(ChannelTags.Event, ChannelTags.Events.Closed).increment()
Expand All @@ -150,7 +150,7 @@ class DbEventHandler(nodeParams: NodeParams) extends Actor with DiagnosticActorL
// spent by the closing transaction.
val capacity = e.commitments.latest.capacity
val fundingTxId = e.commitments.latest.fundingTxId
auditDb.add(ChannelEvent(e.channelId, e.commitments.remoteNodeId, fundingTxId, capacity, e.commitments.localChannelParams.isChannelOpener, !e.commitments.announceChannel, event))
auditDb.add(ChannelEvent(e.channelId, e.commitments.remoteNodeId, fundingTxId, e.commitments.latest.commitmentFormat.toString, capacity, e.commitments.localChannelParams.isChannelOpener, !e.commitments.announceChannel, event.label))
channelsDb.updateChannelMeta(e.channelId, event)

case u: ChannelUpdateParametersChanged =>
Expand Down Expand Up @@ -178,7 +178,7 @@ object DbEventHandler {
def props(nodeParams: NodeParams): Props = Props(new DbEventHandler(nodeParams))

// @formatter:off
case class ChannelEvent(channelId: ByteVector32, remoteNodeId: PublicKey, fundingTxId: TxId, capacity: Satoshi, isChannelOpener: Boolean, isPrivate: Boolean, event: ChannelEvent.EventType)
case class ChannelEvent(channelId: ByteVector32, remoteNodeId: PublicKey, fundingTxId: TxId, channelType: String, capacity: Satoshi, isChannelOpener: Boolean, isPrivate: Boolean, event: String, timestamp: TimestampMilli = TimestampMilli.now())
object ChannelEvent {
sealed trait EventType { def label: String }
object EventType {
Expand All @@ -190,12 +190,12 @@ object DbEventHandler {
object PaymentReceived extends EventType { override def label: String = "received" }
case class Closed(closingType: ClosingType) extends EventType {
override def label: String = closingType match {
case _: MutualClose => "mutual"
case _: LocalClose => "local"
case _: CurrentRemoteClose => "remote"
case _: NextRemoteClose => "remote"
case _: RecoveryClose => "recovery"
case _: RevokedClose => "revoked"
case _: MutualClose => "mutual-close"
case _: LocalClose => "local-close"
case _: CurrentRemoteClose => "remote-close"
case _: NextRemoteClose => "remote-close"
case _: RecoveryClose => "recovery-close"
case _: RevokedClose => "revoked-close"
}
}
}
Expand Down
Loading