Merge pull request #6146 from yyforyongyu/add-closing-txid

rpcserver: add closing txid in PendingChannels' WaitingCloseChannel
This commit is contained in:
Oliver Gugger authored and GitHub committed 2022-01-12 13:23:16 +01:00
commit ed511bb37f
5 files changed
+1811 -1678

No files matched your search

@@ -78,6 +78,10 @@ connection from the watch-only node.
* [Fix missing label on streamed
transactions](https://github.com/lightningnetwork/lnd/pull/5854).
* [Closing txid is now
exposed](https://github.com/lightningnetwork/lnd/pull/6146) inside
WaitingCloseResp from calling `PendingChannels`.
# Contributors (Alphabetical Order)
* Andras Banki-Horvath
+1640 -1629
View File
File diff suppressed because it is too large. Load diff
+3
View File
@@ -2356,6 +2356,9 @@ message PendingChannelsResponse {
this point.
*/
Commitments commitments = 3;
// The transaction id of the closing transaction
string closing_txid = 4;
}
message Commitments {
+4
View File
@@ -2808,6 +2808,10 @@
"commitments": {
"$ref": "#/definitions/PendingChannelsResponseCommitments",
"description": "A list of valid commitment transactions. Any of these can confirm at\nthis point."
},
"closing_txid": {
"type": "string",
"title": "The transaction id of the closing transaction"
}
}
},
+160 -49
View File
@@ -3190,38 +3190,27 @@ func (r *rpcServer) ChannelBalance(ctx context.Context,
}, nil
}
// PendingChannels returns a list of all the channels that are currently
// considered "pending". A channel is pending if it has finished the funding
// workflow and is waiting for confirmations for the funding txn, or is in the
// process of closure, either initiated cooperatively or non-cooperatively.
func (r *rpcServer) PendingChannels(ctx context.Context,
in *lnrpc.PendingChannelsRequest) (*lnrpc.PendingChannelsResponse, error) {
rpcsLog.Debugf("[pendingchannels]")
resp := &lnrpc.PendingChannelsResponse{}
// rpcInitiator returns the correct lnrpc initiator for channels where
// we have a record of the opening channel.
rpcInitiator := func(isInitiator bool) lnrpc.Initiator {
if isInitiator {
return lnrpc.Initiator_INITIATOR_LOCAL
}
return lnrpc.Initiator_INITIATOR_REMOTE
}
type (
pendingOpenChannels []*lnrpc.PendingChannelsResponse_PendingOpenChannel
pendingForceClose []*lnrpc.PendingChannelsResponse_ForceClosedChannel
waitingCloseChannels []*lnrpc.PendingChannelsResponse_WaitingCloseChannel
)
// fetchPendingOpenChannels queries the database for a list of channels that
// have pending open state. The returned result is used in the response of the
// PendingChannels RPC.
func (r *rpcServer) fetchPendingOpenChannels() (pendingOpenChannels, error) {
// First, we'll populate the response with all the channels that are
// soon to be opened. We can easily fetch this data from the database
// and map the db struct to the proto response.
pendingOpenChannels, err := r.server.chanStateDB.FetchPendingChannels()
channels, err := r.server.chanStateDB.FetchPendingChannels()
if err != nil {
rpcsLog.Errorf("unable to fetch pending channels: %v", err)
return nil, err
}
resp.PendingOpenChannels = make([]*lnrpc.PendingChannelsResponse_PendingOpenChannel,
len(pendingOpenChannels))
for i, pendingChan := range pendingOpenChannels {
result := make(pendingOpenChannels, len(channels))
for i, pendingChan := range channels {
pub := pendingChan.IdentityPub.SerializeCompressed()
// As this is required for display purposes, we'll calculate
@@ -3236,7 +3225,7 @@ func (r *rpcServer) PendingChannels(ctx context.Context,
commitBaseWeight := blockchain.GetTransactionWeight(utx)
commitWeight := commitBaseWeight + input.WitnessCommitmentTxWeight
resp.PendingOpenChannels[i] = &lnrpc.PendingChannelsResponse_PendingOpenChannel{
result[i] = &lnrpc.PendingChannelsResponse_PendingOpenChannel{
Channel: &lnrpc.PendingChannelsResponse_PendingChannel{
RemoteNodePub: hex.EncodeToString(pub),
ChannelPoint: pendingChan.FundingOutpoint.String(),
@@ -3255,19 +3244,32 @@ func (r *rpcServer) PendingChannels(ctx context.Context,
}
}
return result, nil
}
// fetchPendingForceCloseChannels queries the database for a list of channels
// that have their closing transactions confirmed but not fully resolved yet.
// The returned result is used in the response of the PendingChannels RPC.
func (r *rpcServer) fetchPendingForceCloseChannels() (pendingForceClose,
int64, error) {
_, currentHeight, err := r.server.cc.ChainIO.GetBestBlock()
if err != nil {
return nil, err
return nil, 0, err
}
// Next, we'll examine the channels that are soon to be closed so we
// can populate these fields within the response.
pendingCloseChannels, err := r.server.chanStateDB.FetchClosedChannels(true)
channels, err := r.server.chanStateDB.FetchClosedChannels(true)
if err != nil {
rpcsLog.Errorf("unable to fetch closed channels: %v", err)
return nil, err
return nil, 0, err
}
for _, pendingClose := range pendingCloseChannels {
result := make(pendingForceClose, 0)
limboBalance := int64(0)
for _, pendingClose := range channels {
// First construct the channel struct itself, this will be
// needed regardless of how this channel was closed.
pub := pendingClose.RemotePub.SerializeCompressed()
@@ -3313,7 +3315,7 @@ func (r *rpcServer) PendingChannels(ctx context.Context,
rpcsLog.Errorf("unable to load forwarding "+
"packages for channel:%s, %v",
historical.ShortChannelID, err)
return nil, err
return nil, 0, err
}
channel.NumForwardingPackages = int64(len(fwdPkgs))
@@ -3324,7 +3326,7 @@ func (r *rpcServer) PendingChannels(ctx context.Context,
// If the error is non-nil, and not due to older versions of lnd
// not persisting historical channels, return it.
default:
return nil, err
return nil, 0, err
}
closeTXID := pendingClose.ClosingTXID.String()
@@ -3359,36 +3361,83 @@ func (r *rpcServer) PendingChannels(ctx context.Context,
&chanPoint, currentHeight, forceClose,
)
if err != nil {
return nil, err
return nil, 0, err
}
err = r.arbitratorPopulateForceCloseResp(
&chanPoint, currentHeight, forceClose,
)
if err != nil {
return nil, err
return nil, 0, err
}
resp.TotalLimboBalance += forceClose.LimboBalance
resp.PendingForceClosingChannels = append(
resp.PendingForceClosingChannels,
forceClose,
)
limboBalance += forceClose.LimboBalance
result = append(result, forceClose)
}
}
return result, limboBalance, nil
}
// fetchWaitingCloseChannels queries the database for a list of channels
// that have their closing transactions broadcast but not confirmed yet.
// The returned result is used in the response of the PendingChannels RPC.
func (r *rpcServer) fetchWaitingCloseChannels() (waitingCloseChannels,
int64, error) {
// We'll also fetch all channels that are open, but have had their
// commitment broadcasted, meaning they are waiting for the closing
// transaction to confirm.
waitingCloseChans, err := r.server.chanStateDB.FetchWaitingCloseChannels()
channels, err := r.server.chanStateDB.FetchWaitingCloseChannels()
if err != nil {
rpcsLog.Errorf("unable to fetch channels waiting close: %v",
err)
return nil, err
return nil, 0, err
}
for _, waitingClose := range waitingCloseChans {
result := make(waitingCloseChannels, 0)
limboBalance := int64(0)
// getClosingTx is a helper closure that tries to find the closing txid
// of a given waiting close channel. Notice that if the remote closes
// the channel, we may not have the closing txid.
getClosingTx := func(c *channeldb.OpenChannel) (string, error) {
var (
tx *wire.MsgTx
err error
)
// First, we try to locate the force closing txid. If not
// found, we will then try to find its coop closing txid.
tx, err = c.BroadcastedCommitment()
if err == nil {
return tx.TxHash().String(), nil
}
// If the error returned is not ErrNoCloseTx, something
// unexpected happened and we will return the error.
if err != channeldb.ErrNoCloseTx {
return "", err
}
// Otherwise, we continue to locate its coop closing txid.
tx, err = c.BroadcastedCooperative()
if err == nil {
return tx.TxHash().String(), nil
}
// Return the error if it's not ErrNoCloseTx.
if err != channeldb.ErrNoCloseTx {
return "", err
}
// Otherwise return an empty txid. This can happen if the
// remote broadcast the closing txid and we haven't recorded it
// yet.
return "", nil
}
for _, waitingClose := range channels {
pub := waitingClose.IdentityPub.SerializeCompressed()
chanPoint := waitingClose.FundingOutpoint
@@ -3426,7 +3475,7 @@ func (r *rpcServer) PendingChannels(ctx context.Context,
// An unexpected error occurred.
case err != nil:
return nil, err
return nil, 0, err
// There is a pending remote commit. Set its hash in the
// response.
@@ -3443,7 +3492,18 @@ func (r *rpcServer) PendingChannels(ctx context.Context,
rpcsLog.Errorf("unable to load forwarding packages "+
"for channel:%s, %v",
waitingClose.ShortChannelID, err)
return nil, err
return nil, 0, err
}
// Get the closing txid.
// NOTE: the closing txid could be empty here if it's the
// remote broadcasted the closing tx.
closingTxid, err := getClosingTx(waitingClose)
if err != nil {
rpcsLog.Errorf("unable to find closing txid for "+
"channel:%s, %v",
waitingClose.ShortChannelID, err)
return nil, 0, err
}
channel := &lnrpc.PendingChannelsResponse_PendingChannel{
@@ -3464,17 +3524,58 @@ func (r *rpcServer) PendingChannels(ctx context.Context,
Channel: channel,
LimboBalance: channel.LocalBalance,
Commitments: &commitments,
ClosingTxid: closingTxid,
}
// A close tx has been broadcasted, all our balance will be in
// limbo until it confirms.
resp.WaitingCloseChannels = append(
resp.WaitingCloseChannels, waitingCloseResp,
)
resp.TotalLimboBalance += channel.LocalBalance
result = append(result, waitingCloseResp)
limboBalance += channel.LocalBalance
}
return result, limboBalance, nil
}
// PendingChannels returns a list of all the channels that are currently
// considered "pending". A channel is pending if it has finished the funding
// workflow and is waiting for confirmations for the funding txn, or is in the
// process of closure, either initiated cooperatively or non-cooperatively.
func (r *rpcServer) PendingChannels(ctx context.Context,
in *lnrpc.PendingChannelsRequest) (
*lnrpc.PendingChannelsResponse, error) {
rpcsLog.Debugf("[pendingchannels]")
resp := &lnrpc.PendingChannelsResponse{}
// First, we find all the channels that will soon be opened.
pendingOpenChannels, err := r.fetchPendingOpenChannels()
if err != nil {
return nil, err
}
resp.PendingOpenChannels = pendingOpenChannels
// Second, we fetch all channels that considered pending force closing.
// This means the channels here have their closing transactions
// confirmed but not considered fully resolved yet. For instance, they
// may have a second level HTLCs to be resolved onchain.
pendingCloseChannels, limbo, err := r.fetchPendingForceCloseChannels()
if err != nil {
return nil, err
}
resp.PendingForceClosingChannels = pendingCloseChannels
resp.TotalLimboBalance = limbo
// Third, we fetch all channels that are open, but have had their
// commitment broadcasted, meaning they are waiting for the closing
// transaction to confirm.
waitingCloseChannels, limbo, err := r.fetchWaitingCloseChannels()
if err != nil {
return nil, err
}
resp.WaitingCloseChannels = waitingCloseChannels
resp.TotalLimboBalance += limbo
return resp, nil
}
@@ -7422,3 +7523,13 @@ func (r *rpcServer) SubscribeCustomMessages(req *lnrpc.SubscribeCustomMessagesRe
}
}
}
// rpcInitiator returns the correct lnrpc initiator for channels where we have
// a record of the opening channel.
func rpcInitiator(isInitiator bool) lnrpc.Initiator {
if isInitiator {
return lnrpc.Initiator_INITIATOR_LOCAL
}
return lnrpc.Initiator_INITIATOR_REMOTE
}