13157733617

Committed 05 Feb 2025 12:49PM UTC coverage: 57.712% (-1.1%) from 58.82%

Build # 13157733617

Build Type

Pull #9447

github

Committed by

yyforyongyu

Commit Message

sweep: rename methods for clarity

We now rename "third party" to "unknown" as the inputs can be spent via
an older sweeping tx, a third party (anchor), or a remote party (pin).
In fee bumper we don't have the info to distinguish the above cases, and
leave them to be further handled by the sweeper as it has more context.

Pull Request Pull Request #9447: sweep: start tracking input spending status in the fee bumper

Run Details

83 of 87 new or added lines in 2 files covered. (95.4%)

19472 existing lines in 252 files now uncovered.

103634 of 179570 relevant lines covered (57.71%)

24840.31 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

81.48

/funding/batch.go

package funding

import (
        "bytes"
        "context"
        "crypto/rand"
        "encoding/base64"
        "errors"
        "fmt"

        "github.com/btcsuite/btcd/btcutil"
        "github.com/btcsuite/btcd/btcutil/psbt"
        "github.com/btcsuite/btcd/chaincfg"
        "github.com/btcsuite/btcd/chaincfg/chainhash"
        "github.com/btcsuite/btcd/wire"
        "github.com/lightningnetwork/lnd/labels"
        "github.com/lightningnetwork/lnd/lnrpc"
        "github.com/lightningnetwork/lnd/lnrpc/walletrpc"
        "github.com/lightningnetwork/lnd/lnwallet/chanfunding"
        "golang.org/x/sync/errgroup"
)

var (
        // errShuttingDown is the error that is returned if a signal on the
        // quit channel is received which means the whole server is shutting
        // down.
        errShuttingDown = errors.New("shutting down")

        // emptyChannelID is a channel ID that consists of all zeros.
        emptyChannelID = [32]byte{}
)

// batchChannel is a struct that keeps track of a single channel's state within
// the batch funding process.
type batchChannel struct {
        fundingReq    *InitFundingMsg
        pendingChanID [32]byte
        updateChan    chan *lnrpc.OpenStatusUpdate
        errChan       chan error
        fundingAddr   string
        chanPoint     *wire.OutPoint
        isPending     bool
}

// processPsbtUpdate processes the first channel update message that is sent
// once the initial part of the negotiation has completed and the funding output
// (and therefore address) is known.
func (c *batchChannel) processPsbtUpdate(u *lnrpc.OpenStatusUpdate) error {
        psbtUpdate := u.GetPsbtFund()
        if psbtUpdate == nil {
                return fmt.Errorf("got unexpected channel update %v", u.Update)
        }

        if psbtUpdate.FundingAmount != int64(c.fundingReq.LocalFundingAmt) {
                return fmt.Errorf("got unexpected funding amount %d, wanted "+
                        "%d", psbtUpdate.FundingAmount,
                        c.fundingReq.LocalFundingAmt)
        }

        c.fundingAddr = psbtUpdate.FundingAddress

        return nil
}

// processPendingUpdate is the second channel update message that is sent once
// the negotiation with the peer has completed and the channel is now pending.
func (c *batchChannel) processPendingUpdate(u *lnrpc.OpenStatusUpdate) error {
        pendingUpd := u.GetChanPending()
        if pendingUpd == nil {
                return fmt.Errorf("got unexpected channel update %v", u.Update)
        }

        hash, err := chainhash.NewHash(pendingUpd.Txid)
        if err != nil {
                return fmt.Errorf("could not parse outpoint TX hash: %w", err)
        }

        c.chanPoint = &wire.OutPoint{
                Index: pendingUpd.OutputIndex,
                Hash:  *hash,
        }
        c.isPending = true

        return nil
}

// RequestParser is a function that parses an incoming RPC request into the
// internal funding initialization message.
type RequestParser func(*lnrpc.OpenChannelRequest) (*InitFundingMsg, error)

// ChannelOpener is a function that kicks off the initial channel open
// negotiation with the peer.
type ChannelOpener func(*InitFundingMsg) (chan *lnrpc.OpenStatusUpdate,
        chan error)

// ChannelAbandoner is a function that can abandon a channel in the local
// database, graph and arbitrator state.
type ChannelAbandoner func(*wire.OutPoint) error

// WalletKitServer is a local interface that abstracts away the methods we need
// from the wallet kit sub server instance.
type WalletKitServer interface {
        // FundPsbt creates a fully populated PSBT that contains enough inputs
        // to fund the outputs specified in the template.
        FundPsbt(context.Context,
                *walletrpc.FundPsbtRequest) (*walletrpc.FundPsbtResponse, error)

        // FinalizePsbt expects a partial transaction with all inputs and
        // outputs fully declared and tries to sign all inputs that belong to
        // the wallet.
        FinalizePsbt(context.Context,
                *walletrpc.FinalizePsbtRequest) (*walletrpc.FinalizePsbtResponse,
                error)

        // ReleaseOutput unlocks an output, allowing it to be available for coin
        // selection if it remains unspent. The ID should match the one used to
        // originally lock the output.
        ReleaseOutput(context.Context,
                *walletrpc.ReleaseOutputRequest) (*walletrpc.ReleaseOutputResponse,
                error)
}

// Wallet is a local interface that abstracts away the methods we need from the
// internal lightning wallet instance.
type Wallet interface {
        // PsbtFundingVerify looks up a previously registered funding intent by
        // its pending channel ID and tries to advance the state machine by
        // verifying the passed PSBT.
        PsbtFundingVerify([32]byte, *psbt.Packet, bool) error

        // PsbtFundingFinalize looks up a previously registered funding intent
        // by its pending channel ID and tries to advance the state machine by
        // finalizing the passed PSBT.
        PsbtFundingFinalize([32]byte, *psbt.Packet, *wire.MsgTx) error

        // PublishTransaction performs cursory validation (dust checks, etc),
        // then finally broadcasts the passed transaction to the Bitcoin
        // network.
        PublishTransaction(*wire.MsgTx, string) error

        // CancelFundingIntent allows a caller to cancel a previously registered
        // funding intent. If no intent was found, then an error will be
        // returned.
        CancelFundingIntent([32]byte) error
}

// BatchConfig is the configuration for executing a single batch transaction for
// opening multiple channels atomically.
type BatchConfig struct {
        // RequestParser is the function that parses an incoming RPC request
        // into the internal funding initialization message.
        RequestParser RequestParser

        // ChannelOpener is the function that kicks off the initial channel open
        // negotiation with the peer.
        ChannelOpener ChannelOpener

        // ChannelAbandoner is the function that can abandon a channel in the
        // local database, graph and arbitrator state.
        ChannelAbandoner ChannelAbandoner

        // WalletKitServer is an instance of the wallet kit sub server that can
        // handle PSBT funding and finalization.
        WalletKitServer WalletKitServer

        // Wallet is an instance of the internal lightning wallet.
        Wallet Wallet

        // NetParams contains the current bitcoin network parameters.
        NetParams *chaincfg.Params

        // Quit is the channel that is selected on to recognize if the main
        // server is shutting down.
        Quit chan struct{}
}

// Batcher is a type that can be used to perform an atomic funding of multiple
// channels within a single on-chain transaction.
type Batcher struct {
        cfg *BatchConfig

        channels    []*batchChannel
        lockedUTXOs []*walletrpc.UtxoLease

        didPublish bool
}

// NewBatcher returns a new batch channel funding helper.
func NewBatcher(cfg *BatchConfig) *Batcher {
        return &Batcher{
                cfg: cfg,
        }
}

// BatchFund starts the atomic batch channel funding process.
//
// NOTE: This method should only be called once per instance.
func (b *Batcher) BatchFund(ctx context.Context,
        req *lnrpc.BatchOpenChannelRequest) ([]*lnrpc.PendingUpdate, error) {

        label, err := labels.ValidateAPI(req.Label)
        if err != nil {
                return nil, err
        }

        // Parse and validate each individual channel.
        b.channels = make([]*batchChannel, 0, len(req.Channels))
        for idx, rpcChannel := range req.Channels {
                // If the user specifies a channel ID, it must be exactly 32
                // bytes long.
                if len(rpcChannel.PendingChanId) > 0 &&
                        len(rpcChannel.PendingChanId) != 32 {

                        return nil, fmt.Errorf("invalid temp chan ID %x",
                                rpcChannel.PendingChanId)
                }

                var pendingChanID [32]byte
                if len(rpcChannel.PendingChanId) == 32 {
                        copy(pendingChanID[:], rpcChannel.PendingChanId)

                        // Don't allow the user to be clever by just setting an
                        // all zero channel ID, we need a "real" value here.
                        if pendingChanID == emptyChannelID {
                                return nil, fmt.Errorf("invalid empty temp " +
                                        "chan ID")
                        }
                } else if _, err := rand.Read(pendingChanID[:]); err != nil {
                        return nil, fmt.Errorf("error making temp chan ID: %w",
                                err)
                }

                //nolint:ll
                fundingReq, err := b.cfg.RequestParser(&lnrpc.OpenChannelRequest{
                        SatPerVbyte:                uint64(req.SatPerVbyte),
                        TargetConf:                 req.TargetConf,
                        MinConfs:                   req.MinConfs,
                        SpendUnconfirmed:           req.SpendUnconfirmed,
                        NodePubkey:                 rpcChannel.NodePubkey,
                        LocalFundingAmount:         rpcChannel.LocalFundingAmount,
                        PushSat:                    rpcChannel.PushSat,
                        Private:                    rpcChannel.Private,
                        MinHtlcMsat:                rpcChannel.MinHtlcMsat,
                        RemoteCsvDelay:             rpcChannel.RemoteCsvDelay,
                        CloseAddress:               rpcChannel.CloseAddress,
                        RemoteMaxValueInFlightMsat: rpcChannel.RemoteMaxValueInFlightMsat,
                        RemoteMaxHtlcs:             rpcChannel.RemoteMaxHtlcs,
                        MaxLocalCsv:                rpcChannel.MaxLocalCsv,
                        CommitmentType:             rpcChannel.CommitmentType,
                        ZeroConf:                   rpcChannel.ZeroConf,
                        ScidAlias:                  rpcChannel.ScidAlias,
                        BaseFee:                    rpcChannel.BaseFee,
                        FeeRate:                    rpcChannel.FeeRate,
                        UseBaseFee:                 rpcChannel.UseBaseFee,
                        UseFeeRate:                 rpcChannel.UseFeeRate,
                        RemoteChanReserveSat:       rpcChannel.RemoteChanReserveSat,
                        Memo:                       rpcChannel.Memo,
                        FundingShim: &lnrpc.FundingShim{
                                Shim: &lnrpc.FundingShim_PsbtShim{
                                        PsbtShim: &lnrpc.PsbtShim{
                                                PendingChanId: pendingChanID[:],
                                                NoPublish:     true,
                                        },
                                },
                        },
                })
                if err != nil {
                        return nil, fmt.Errorf("error parsing channel %d: %w",
                                idx, err)
                }

                // Prepare the stuff that we'll need for the internal PSBT
                // funding.
                fundingReq.PendingChanID = pendingChanID
                fundingReq.ChanFunder = chanfunding.NewPsbtAssembler(
                        btcutil.Amount(rpcChannel.LocalFundingAmount), nil,
                        b.cfg.NetParams, false,
                )

                b.channels = append(b.channels, &batchChannel{
                        pendingChanID: pendingChanID,
                        fundingReq:    fundingReq,
                })
        }

        // From this point on we can fail for any of the channels and for any
        // number of reasons. This deferred function makes sure that the full
        // operation is actually atomic: We either succeed and publish a
        // transaction for the full batch or we clean up everything.
        defer b.cleanup(ctx)

        // Now that we know the user input is sane, we need to kick off the
        // channel funding negotiation with the peers. Because we specified a
        // PSBT assembler, we'll get a special response in the channel once the
        // funding output script is known (which we need to craft the TX).
        eg := &errgroup.Group{}
        for _, channel := range b.channels {
                channel.updateChan, channel.errChan = b.cfg.ChannelOpener(
                        channel.fundingReq,
                )

                // Launch a goroutine that waits for the initial response on
                // either the update or error chan.
                channel := channel
                eg.Go(func() error {
                        return b.waitForUpdate(channel, true)
                })
        }

        // Wait for all goroutines to report back. Any error at this stage means
        // we need to abort.
        if err := eg.Wait(); err != nil {
                return nil, fmt.Errorf("error batch opening channel, initial "+
                        "negotiation failed: %v", err)
        }

        // We can now assemble all outputs that we're going to give to the PSBT
        // funding method of the wallet kit server.
        txTemplate := &walletrpc.TxTemplate{
                Outputs: make(map[string]uint64),
        }
        for _, channel := range b.channels {
                txTemplate.Outputs[channel.fundingAddr] = uint64(
                        channel.fundingReq.LocalFundingAmt,
                )
        }

        // Great, we've now started the channel negotiation successfully with
        // all peers. This means we know the channel outputs for all channels
        // and can craft our PSBT now. We take the fee rate and min conf
        // settings from the first request as all of them should be equal
        // anyway.
        firstReq := b.channels[0].fundingReq
        feeRateSatPerVByte := firstReq.FundingFeePerKw.FeePerVByte()
        changeType := walletrpc.ChangeAddressType_CHANGE_ADDRESS_TYPE_P2TR
        fundPsbtReq := &walletrpc.FundPsbtRequest{
                Template: &walletrpc.FundPsbtRequest_Raw{
                        Raw: txTemplate,
                },
                Fees: &walletrpc.FundPsbtRequest_SatPerVbyte{
                        SatPerVbyte: uint64(feeRateSatPerVByte),
                },
                MinConfs:              firstReq.MinConfs,
                SpendUnconfirmed:      firstReq.MinConfs == 0,
                ChangeType:            changeType,
                CoinSelectionStrategy: req.CoinSelectionStrategy,
        }
        fundPsbtResp, err := b.cfg.WalletKitServer.FundPsbt(ctx, fundPsbtReq)
        if err != nil {
                return nil, fmt.Errorf("error funding PSBT for batch channel "+
                        "open: %v", err)
        }

        // Funding was successful. This means there are some UTXOs that are now
        // locked for us. We need to make sure we release them if we don't
        // complete the publish process.
        b.lockedUTXOs = fundPsbtResp.LockedUtxos

        // Parse and log the funded PSBT for debugging purposes.
        unsignedPacket, err := psbt.NewFromRawBytes(
                bytes.NewReader(fundPsbtResp.FundedPsbt), false,
        )
        if err != nil {
                return nil, fmt.Errorf("error parsing funded PSBT for batch "+
                        "channel open: %v", err)
        }
        log.Tracef("[batchopenchannel] funded PSBT: %s",
                base64.StdEncoding.EncodeToString(fundPsbtResp.FundedPsbt))

        // With the funded PSBT we can now advance the funding state machine of
        // each of the channels.
        for _, channel := range b.channels {
                err = b.cfg.Wallet.PsbtFundingVerify(
                        channel.pendingChanID, unsignedPacket, false,
                )
                if err != nil {
                        return nil, fmt.Errorf("error verifying PSBT: %w", err)
                }
        }

        // The funded PSBT was accepted by each of the assemblers, let's now
        // sign/finalize it.
        finalizePsbtResp, err := b.cfg.WalletKitServer.FinalizePsbt(
                ctx, &walletrpc.FinalizePsbtRequest{
                        FundedPsbt: fundPsbtResp.FundedPsbt,
                },
        )
        if err != nil {
                return nil, fmt.Errorf("error finalizing PSBT for batch "+
                        "channel open: %v", err)
        }
        finalTx := &wire.MsgTx{}
        txReader := bytes.NewReader(finalizePsbtResp.RawFinalTx)
        if err := finalTx.Deserialize(txReader); err != nil {
                return nil, fmt.Errorf("error parsing signed raw TX: %w", err)
        }
        log.Tracef("[batchopenchannel] signed PSBT: %s",
                base64.StdEncoding.EncodeToString(finalizePsbtResp.SignedPsbt))

        // Advance the funding state machine of each of the channels a last time
        // to complete the negotiation with the now signed funding TX.
        for _, channel := range b.channels {
                err = b.cfg.Wallet.PsbtFundingFinalize(
                        channel.pendingChanID, nil, finalTx,
                )
                if err != nil {
                        return nil, fmt.Errorf("error finalizing PSBT: %w", err)
                }
        }

        // Now every channel should be ready for the funding transaction to be
        // broadcast. Let's wait for the updates that actually confirm this
        // state.
        eg = &errgroup.Group{}
        for _, channel := range b.channels {
                // Launch another goroutine that waits for the channel pending
                // response on the update chan.
                channel := channel
                eg.Go(func() error {
                        return b.waitForUpdate(channel, false)
                })
        }

        // Wait for all updates and make sure we're still good to proceed.
        if err := eg.Wait(); err != nil {
                return nil, fmt.Errorf("error batch opening channel, final "+
                        "negotiation failed: %v", err)
        }

        // Great, we're now finally ready to publish the transaction.
        err = b.cfg.Wallet.PublishTransaction(finalTx, label)
        if err != nil {
                return nil, fmt.Errorf("error publishing final batch "+
                        "transaction: %v", err)
        }
        b.didPublish = true

        rpcPoints := make([]*lnrpc.PendingUpdate, len(b.channels))
        for idx, channel := range b.channels {
                rpcPoints[idx] = &lnrpc.PendingUpdate{
                        Txid:        channel.chanPoint.Hash.CloneBytes(),
                        OutputIndex: channel.chanPoint.Index,
                }
        }

        return rpcPoints, nil
}

// waitForUpdate waits for an incoming channel update (or error) for a single
// channel.
//
// NOTE: Must be called in a goroutine as this blocks until an update or error
// is received.
func (b *Batcher) waitForUpdate(channel *batchChannel, firstUpdate bool) error {
        select {
        // If an error occurs then immediately return the error to the client.
        case err := <-channel.errChan:
                log.Errorf("unable to open channel to NodeKey(%x): %v",
                        channel.fundingReq.TargetPubkey.SerializeCompressed(),
                        err)
                return err

        // Otherwise, wait for the next channel update. The first update sent
        // must be the signal to start the PSBT funding in our case since we
        // specified a PSBT shim. The second update will be the signal that the
        // channel is now pending.
        case fundingUpdate := <-channel.updateChan:
                log.Tracef("[batchopenchannel] received update: %v",
                        fundingUpdate)

                // Depending on what update we were waiting for the batch
                // channel knows what to do with it.
                if firstUpdate {
                        return channel.processPsbtUpdate(fundingUpdate)
                }

                return channel.processPendingUpdate(fundingUpdate)

        case <-b.cfg.Quit:
                return errShuttingDown
        }
}

// cleanup tries to remove any pending state or UTXO locks in case we had to
// abort before finalizing and publishing the funding transaction.
func (b *Batcher) cleanup(ctx context.Context) {
        // Did we publish a transaction? Then there's nothing to clean up since
        // we succeeded.
        if b.didPublish {
                return
        }

        // Make sure the error message doesn't sound too scary. These might be
        // logged quite frequently depending on where exactly things were
        // aborted. We could just not log any cleanup errors though it might be
        // helpful to debug things if something doesn't go as expected.
        const errMsgTpl = "Attempted to clean up after failed batch channel " +
                "open but could not %s: %v"

        // If we failed, we clean up in reverse order. First, let's unlock the
        // leased outputs.
        for _, lockedUTXO := range b.lockedUTXOs {
                rpcOP := &lnrpc.OutPoint{
                        OutputIndex: lockedUTXO.Outpoint.OutputIndex,
                        TxidBytes:   lockedUTXO.Outpoint.TxidBytes,
                        TxidStr:     lockedUTXO.Outpoint.TxidStr,
                }
                _, err := b.cfg.WalletKitServer.ReleaseOutput(
                        ctx, &walletrpc.ReleaseOutputRequest{
                                Id:       lockedUTXO.Id,
                                Outpoint: rpcOP,
                        },
                )
                if err != nil {
                        log.Debugf(errMsgTpl, "release locked output "+
                                lockedUTXO.Outpoint.String(), err)
                }
        }

        // Then go through all channels that ever got into a pending state and
        // remove the pending channel by abandoning them.
        for _, channel := range b.channels {
                if !channel.isPending {
                        continue
                }

                err := b.cfg.ChannelAbandoner(channel.chanPoint)
                if err != nil {
                        log.Debugf(errMsgTpl, "abandon pending open channel",
                                err)
                }
        }

        // And finally clean up the funding shim for each channel that didn't
        // make it into a pending state.
        for _, channel := range b.channels {
                if channel.isPending {
                        continue
                }

                err := b.cfg.Wallet.CancelFundingIntent(channel.pendingChanID)
                if err != nil {
                        log.Debugf(errMsgTpl, "cancel funding shim", err)
                }
        }
}

1	package funding
2
3	import (
4	"bytes"
5	"context"
6	"crypto/rand"
7	"encoding/base64"
8	"errors"
9	"fmt"
10
11	"github.com/btcsuite/btcd/btcutil"
12	"github.com/btcsuite/btcd/btcutil/psbt"
13	"github.com/btcsuite/btcd/chaincfg"
14	"github.com/btcsuite/btcd/chaincfg/chainhash"
15	"github.com/btcsuite/btcd/wire"
16	"github.com/lightningnetwork/lnd/labels"
17	"github.com/lightningnetwork/lnd/lnrpc"
18	"github.com/lightningnetwork/lnd/lnrpc/walletrpc"
19	"github.com/lightningnetwork/lnd/lnwallet/chanfunding"
20	"golang.org/x/sync/errgroup"
21	)
22
23	var (
24	// errShuttingDown is the error that is returned if a signal on the
25	// quit channel is received which means the whole server is shutting
26	// down.
27	errShuttingDown = errors.New("shutting down")
28
29	// emptyChannelID is a channel ID that consists of all zeros.
30	emptyChannelID = [32]byte{}
31	)
32
33	// batchChannel is a struct that keeps track of a single channel's state within
34	// the batch funding process.
35	type batchChannel struct {
36	fundingReq *InitFundingMsg
37	pendingChanID [32]byte
38	updateChan chan *lnrpc.OpenStatusUpdate
39	errChan chan error
40	fundingAddr string
41	chanPoint *wire.OutPoint
42	isPending bool
43	}
44
45	// processPsbtUpdate processes the first channel update message that is sent
46	// once the initial part of the negotiation has completed and the funding output
47	// (and therefore address) is known.
48	func (c batchChannel) processPsbtUpdate(u lnrpc.OpenStatusUpdate) error {	6✔
49	psbtUpdate := u.GetPsbtFund()	6✔
50	if psbtUpdate == nil {	6✔
51	return fmt.Errorf("got unexpected channel update %v", u.Update)	×
52	}	×
53
54	if psbtUpdate.FundingAmount != int64(c.fundingReq.LocalFundingAmt) {	6✔
55	return fmt.Errorf("got unexpected funding amount %d, wanted "+	×
56	"%d", psbtUpdate.FundingAmount,	×
57	c.fundingReq.LocalFundingAmt)	×
58	}	×
59
60	c.fundingAddr = psbtUpdate.FundingAddress	6✔
61		6✔
62	return nil	6✔
63	}
64
65	// processPendingUpdate is the second channel update message that is sent once
66	// the negotiation with the peer has completed and the channel is now pending.
67	func (c batchChannel) processPendingUpdate(u lnrpc.OpenStatusUpdate) error {	5✔
68	pendingUpd := u.GetChanPending()	5✔
69	if pendingUpd == nil {	5✔
70	return fmt.Errorf("got unexpected channel update %v", u.Update)	×
71	}	×
72
73	hash, err := chainhash.NewHash(pendingUpd.Txid)	5✔
74	if err != nil {	5✔
75	return fmt.Errorf("could not parse outpoint TX hash: %w", err)	×
76	}	×
77
78	c.chanPoint = &wire.OutPoint{	5✔
79	Index: pendingUpd.OutputIndex,	5✔
80	Hash: *hash,	5✔
81	}	5✔
82	c.isPending = true	5✔
83		5✔
84	return nil	5✔
85	}
86
87	// RequestParser is a function that parses an incoming RPC request into the
88	// internal funding initialization message.
89	type RequestParser func(lnrpc.OpenChannelRequest) (InitFundingMsg, error)
90
91	// ChannelOpener is a function that kicks off the initial channel open
92	// negotiation with the peer.
93	type ChannelOpener func(InitFundingMsg) (chan lnrpc.OpenStatusUpdate,
94	chan error)
95
96	// ChannelAbandoner is a function that can abandon a channel in the local
97	// database, graph and arbitrator state.
98	type ChannelAbandoner func(*wire.OutPoint) error
99
100	// WalletKitServer is a local interface that abstracts away the methods we need
101	// from the wallet kit sub server instance.
102	type WalletKitServer interface {
103	// FundPsbt creates a fully populated PSBT that contains enough inputs
104	// to fund the outputs specified in the template.
105	FundPsbt(context.Context,
106	walletrpc.FundPsbtRequest) (walletrpc.FundPsbtResponse, error)
107
108	// FinalizePsbt expects a partial transaction with all inputs and
109	// outputs fully declared and tries to sign all inputs that belong to
110	// the wallet.
111	FinalizePsbt(context.Context,
112	walletrpc.FinalizePsbtRequest) (walletrpc.FinalizePsbtResponse,
113	error)
114
115	// ReleaseOutput unlocks an output, allowing it to be available for coin
116	// selection if it remains unspent. The ID should match the one used to
117	// originally lock the output.
118	ReleaseOutput(context.Context,
119	walletrpc.ReleaseOutputRequest) (walletrpc.ReleaseOutputResponse,
120	error)
121	}
122
123	// Wallet is a local interface that abstracts away the methods we need from the
124	// internal lightning wallet instance.
125	type Wallet interface {
126	// PsbtFundingVerify looks up a previously registered funding intent by
127	// its pending channel ID and tries to advance the state machine by
128	// verifying the passed PSBT.
129	PsbtFundingVerify([32]byte, *psbt.Packet, bool) error
130
131	// PsbtFundingFinalize looks up a previously registered funding intent
132	// by its pending channel ID and tries to advance the state machine by
133	// finalizing the passed PSBT.
134	PsbtFundingFinalize([32]byte, psbt.Packet, wire.MsgTx) error
135
136	// PublishTransaction performs cursory validation (dust checks, etc),
137	// then finally broadcasts the passed transaction to the Bitcoin
138	// network.
139	PublishTransaction(*wire.MsgTx, string) error
140
141	// CancelFundingIntent allows a caller to cancel a previously registered
142	// funding intent. If no intent was found, then an error will be
143	// returned.
144	CancelFundingIntent([32]byte) error
145	}
146
147	// BatchConfig is the configuration for executing a single batch transaction for
148	// opening multiple channels atomically.
149	type BatchConfig struct {
150	// RequestParser is the function that parses an incoming RPC request
151	// into the internal funding initialization message.
152	RequestParser RequestParser
153
154	// ChannelOpener is the function that kicks off the initial channel open
155	// negotiation with the peer.
156	ChannelOpener ChannelOpener
157
158	// ChannelAbandoner is the function that can abandon a channel in the
159	// local database, graph and arbitrator state.
160	ChannelAbandoner ChannelAbandoner
161
162	// WalletKitServer is an instance of the wallet kit sub server that can
163	// handle PSBT funding and finalization.
164	WalletKitServer WalletKitServer
165
166	// Wallet is an instance of the internal lightning wallet.
167	Wallet Wallet
168
169	// NetParams contains the current bitcoin network parameters.
170	NetParams *chaincfg.Params
171
172	// Quit is the channel that is selected on to recognize if the main
173	// server is shutting down.
174	Quit chan struct{}
175	}
176
177	// Batcher is a type that can be used to perform an atomic funding of multiple
178	// channels within a single on-chain transaction.
179	type Batcher struct {
180	cfg *BatchConfig
181
182	channels []*batchChannel
183	lockedUTXOs []*walletrpc.UtxoLease
184
185	didPublish bool
186	}
187
188	// NewBatcher returns a new batch channel funding helper.
189	func NewBatcher(cfg BatchConfig) Batcher {	4✔
190	return &Batcher{	4✔
191	cfg: cfg,	4✔
192	}	4✔
193	}	4✔
194
195	// BatchFund starts the atomic batch channel funding process.
196	//
197	// NOTE: This method should only be called once per instance.
198	func (b *Batcher) BatchFund(ctx context.Context,
199	req lnrpc.BatchOpenChannelRequest) ([]lnrpc.PendingUpdate, error) {	4✔
200		4✔
201	label, err := labels.ValidateAPI(req.Label)	4✔
202	if err != nil {	4✔
203	return nil, err	×
204	}	×
205
206	// Parse and validate each individual channel.
207	b.channels = make([]*batchChannel, 0, len(req.Channels))	4✔
208	for idx, rpcChannel := range req.Channels {	12✔
209	// If the user specifies a channel ID, it must be exactly 32	8✔
210	// bytes long.	8✔
211	if len(rpcChannel.PendingChanId) > 0 &&	8✔
212	len(rpcChannel.PendingChanId) != 32 {	8✔
213		×
214	return nil, fmt.Errorf("invalid temp chan ID %x",	×
215	rpcChannel.PendingChanId)	×
216	}	×
217
218	var pendingChanID [32]byte	8✔
219	if len(rpcChannel.PendingChanId) == 32 {	8✔
220	copy(pendingChanID[:], rpcChannel.PendingChanId)	×
221		×
222	// Don't allow the user to be clever by just setting an	×
223	// all zero channel ID, we need a "real" value here.	×
224	if pendingChanID == emptyChannelID {	×
225	return nil, fmt.Errorf("invalid empty temp " +	×
226	"chan ID")	×
227	}	×
228	} else if _, err := rand.Read(pendingChanID[:]); err != nil {	8✔
229	return nil, fmt.Errorf("error making temp chan ID: %w",	×
230	err)	×
231	}	×
232
233	//nolint:ll
234	fundingReq, err := b.cfg.RequestParser(&lnrpc.OpenChannelRequest{	8✔
235	SatPerVbyte: uint64(req.SatPerVbyte),	8✔
236	TargetConf: req.TargetConf,	8✔
237	MinConfs: req.MinConfs,	8✔
238	SpendUnconfirmed: req.SpendUnconfirmed,	8✔
239	NodePubkey: rpcChannel.NodePubkey,	8✔
240	LocalFundingAmount: rpcChannel.LocalFundingAmount,	8✔
241	PushSat: rpcChannel.PushSat,	8✔
242	Private: rpcChannel.Private,	8✔
243	MinHtlcMsat: rpcChannel.MinHtlcMsat,	8✔
244	RemoteCsvDelay: rpcChannel.RemoteCsvDelay,	8✔
245	CloseAddress: rpcChannel.CloseAddress,	8✔
246	RemoteMaxValueInFlightMsat: rpcChannel.RemoteMaxValueInFlightMsat,	8✔
247	RemoteMaxHtlcs: rpcChannel.RemoteMaxHtlcs,	8✔
248	MaxLocalCsv: rpcChannel.MaxLocalCsv,	8✔
249	CommitmentType: rpcChannel.CommitmentType,	8✔
250	ZeroConf: rpcChannel.ZeroConf,	8✔
251	ScidAlias: rpcChannel.ScidAlias,	8✔
252	BaseFee: rpcChannel.BaseFee,	8✔
253	FeeRate: rpcChannel.FeeRate,	8✔
254	UseBaseFee: rpcChannel.UseBaseFee,	8✔
255	UseFeeRate: rpcChannel.UseFeeRate,	8✔
256	RemoteChanReserveSat: rpcChannel.RemoteChanReserveSat,	8✔
257	Memo: rpcChannel.Memo,	8✔
258	FundingShim: &lnrpc.FundingShim{	8✔
259	Shim: &lnrpc.FundingShim_PsbtShim{	8✔
260	PsbtShim: &lnrpc.PsbtShim{	8✔
261	PendingChanId: pendingChanID[:],	8✔
262	NoPublish: true,	8✔
263	},	8✔
264	},	8✔
265	},	8✔
266	})	8✔
267	if err != nil {	8✔
268	return nil, fmt.Errorf("error parsing channel %d: %w",	×
269	idx, err)	×
270	}	×
271
272	// Prepare the stuff that we'll need for the internal PSBT
273	// funding.
274	fundingReq.PendingChanID = pendingChanID	8✔
275	fundingReq.ChanFunder = chanfunding.NewPsbtAssembler(	8✔
276	btcutil.Amount(rpcChannel.LocalFundingAmount), nil,	8✔
277	b.cfg.NetParams, false,	8✔
278	)	8✔
279		8✔
280	b.channels = append(b.channels, &batchChannel{	8✔
281	pendingChanID: pendingChanID,	8✔
282	fundingReq: fundingReq,	8✔
283	})	8✔
284	}
285
286	// From this point on we can fail for any of the channels and for any
287	// number of reasons. This deferred function makes sure that the full
288	// operation is actually atomic: We either succeed and publish a
289	// transaction for the full batch or we clean up everything.
290	defer b.cleanup(ctx)	4✔
291		4✔
292	// Now that we know the user input is sane, we need to kick off the	4✔
293	// channel funding negotiation with the peers. Because we specified a	4✔
294	// PSBT assembler, we'll get a special response in the channel once the	4✔
295	// funding output script is known (which we need to craft the TX).	4✔
296	eg := &errgroup.Group{}	4✔
297	for _, channel := range b.channels {	12✔
298	channel.updateChan, channel.errChan = b.cfg.ChannelOpener(	8✔
299	channel.fundingReq,	8✔
300	)	8✔
301		8✔
302	// Launch a goroutine that waits for the initial response on	8✔
303	// either the update or error chan.	8✔
304	channel := channel	8✔
305	eg.Go(func() error {	16✔
306	return b.waitForUpdate(channel, true)	8✔
307	})	8✔
308	}
309
310	// Wait for all goroutines to report back. Any error at this stage means
311	// we need to abort.
312	if err := eg.Wait(); err != nil {	5✔
313	return nil, fmt.Errorf("error batch opening channel, initial "+	1✔
314	"negotiation failed: %v", err)	1✔
315	}	1✔
316
317	// We can now assemble all outputs that we're going to give to the PSBT
318	// funding method of the wallet kit server.
319	txTemplate := &walletrpc.TxTemplate{	3✔
320	Outputs: make(map[string]uint64),	3✔
321	}	3✔
322	for _, channel := range b.channels {	9✔
323	txTemplate.Outputs[channel.fundingAddr] = uint64(	6✔
324	channel.fundingReq.LocalFundingAmt,	6✔
325	)	6✔
326	}	6✔
327
328	// Great, we've now started the channel negotiation successfully with
329	// all peers. This means we know the channel outputs for all channels
330	// and can craft our PSBT now. We take the fee rate and min conf
331	// settings from the first request as all of them should be equal
332	// anyway.
333	firstReq := b.channels[0].fundingReq	3✔
334	feeRateSatPerVByte := firstReq.FundingFeePerKw.FeePerVByte()	3✔
335	changeType := walletrpc.ChangeAddressType_CHANGE_ADDRESS_TYPE_P2TR	3✔
336	fundPsbtReq := &walletrpc.FundPsbtRequest{	3✔
337	Template: &walletrpc.FundPsbtRequest_Raw{	3✔
338	Raw: txTemplate,	3✔
339	},	3✔
340	Fees: &walletrpc.FundPsbtRequest_SatPerVbyte{	3✔
341	SatPerVbyte: uint64(feeRateSatPerVByte),	3✔
342	},	3✔
343	MinConfs: firstReq.MinConfs,	3✔
344	SpendUnconfirmed: firstReq.MinConfs == 0,	3✔
345	ChangeType: changeType,	3✔
346	CoinSelectionStrategy: req.CoinSelectionStrategy,	3✔
347	}	3✔
348	fundPsbtResp, err := b.cfg.WalletKitServer.FundPsbt(ctx, fundPsbtReq)	3✔
349	if err != nil {	3✔
350	return nil, fmt.Errorf("error funding PSBT for batch channel "+	×
351	"open: %v", err)	×
352	}	×
353
354	// Funding was successful. This means there are some UTXOs that are now
355	// locked for us. We need to make sure we release them if we don't
356	// complete the publish process.
357	b.lockedUTXOs = fundPsbtResp.LockedUtxos	3✔
358		3✔
359	// Parse and log the funded PSBT for debugging purposes.	3✔
360	unsignedPacket, err := psbt.NewFromRawBytes(	3✔
361	bytes.NewReader(fundPsbtResp.FundedPsbt), false,	3✔
362	)	3✔
363	if err != nil {	3✔
364	return nil, fmt.Errorf("error parsing funded PSBT for batch "+	×
365	"channel open: %v", err)	×
366	}	×
367	log.Tracef("[batchopenchannel] funded PSBT: %s",	3✔
368	base64.StdEncoding.EncodeToString(fundPsbtResp.FundedPsbt))	3✔
369		3✔
370	// With the funded PSBT we can now advance the funding state machine of	3✔
371	// each of the channels.	3✔
372	for _, channel := range b.channels {	9✔
373	err = b.cfg.Wallet.PsbtFundingVerify(	6✔
374	channel.pendingChanID, unsignedPacket, false,	6✔
375	)	6✔
376	if err != nil {	6✔
377	return nil, fmt.Errorf("error verifying PSBT: %w", err)	×
378	}	×
379	}
380
381	// The funded PSBT was accepted by each of the assemblers, let's now
382	// sign/finalize it.
383	finalizePsbtResp, err := b.cfg.WalletKitServer.FinalizePsbt(	3✔
384	ctx, &walletrpc.FinalizePsbtRequest{	3✔
385	FundedPsbt: fundPsbtResp.FundedPsbt,	3✔
386	},	3✔
387	)	3✔
388	if err != nil {	3✔
389	return nil, fmt.Errorf("error finalizing PSBT for batch "+	×
390	"channel open: %v", err)	×
391	}	×
392	finalTx := &wire.MsgTx{}	3✔
393	txReader := bytes.NewReader(finalizePsbtResp.RawFinalTx)	3✔
394	if err := finalTx.Deserialize(txReader); err != nil {	3✔
395	return nil, fmt.Errorf("error parsing signed raw TX: %w", err)	×
396	}	×
397	log.Tracef("[batchopenchannel] signed PSBT: %s",	3✔
398	base64.StdEncoding.EncodeToString(finalizePsbtResp.SignedPsbt))	3✔
399		3✔
400	// Advance the funding state machine of each of the channels a last time	3✔
401	// to complete the negotiation with the now signed funding TX.	3✔
402	for _, channel := range b.channels {	9✔
403	err = b.cfg.Wallet.PsbtFundingFinalize(	6✔
404	channel.pendingChanID, nil, finalTx,	6✔
405	)	6✔
406	if err != nil {	6✔
407	return nil, fmt.Errorf("error finalizing PSBT: %w", err)	×
408	}	×
409	}
410
411	// Now every channel should be ready for the funding transaction to be
412	// broadcast. Let's wait for the updates that actually confirm this
413	// state.
414	eg = &errgroup.Group{}	3✔
415	for _, channel := range b.channels {	9✔
416	// Launch another goroutine that waits for the channel pending	6✔
417	// response on the update chan.	6✔
418	channel := channel	6✔
419	eg.Go(func() error {	12✔
420	return b.waitForUpdate(channel, false)	6✔
421	})	6✔
422	}
423
424	// Wait for all updates and make sure we're still good to proceed.
425	if err := eg.Wait(); err != nil {	4✔
426	return nil, fmt.Errorf("error batch opening channel, final "+	1✔
427	"negotiation failed: %v", err)	1✔
428	}	1✔
429
430	// Great, we're now finally ready to publish the transaction.
431	err = b.cfg.Wallet.PublishTransaction(finalTx, label)	2✔
432	if err != nil {	3✔
433	return nil, fmt.Errorf("error publishing final batch "+	1✔
434	"transaction: %v", err)	1✔
435	}	1✔
436	b.didPublish = true	1✔
437		1✔
438	rpcPoints := make([]*lnrpc.PendingUpdate, len(b.channels))	1✔
439	for idx, channel := range b.channels {	3✔
440	rpcPoints[idx] = &lnrpc.PendingUpdate{	2✔
441	Txid: channel.chanPoint.Hash.CloneBytes(),	2✔
442	OutputIndex: channel.chanPoint.Index,	2✔
443	}	2✔
444	}	2✔
445
446	return rpcPoints, nil	1✔
447	}
448
449	// waitForUpdate waits for an incoming channel update (or error) for a single
450	// channel.
451	//
452	// NOTE: Must be called in a goroutine as this blocks until an update or error
453	// is received.
454	func (b Batcher) waitForUpdate(channel batchChannel, firstUpdate bool) error {	14✔
455	select {	14✔
456	// If an error occurs then immediately return the error to the client.
457	case err := <-channel.errChan:	3✔
458	log.Errorf("unable to open channel to NodeKey(%x): %v",	3✔
459	channel.fundingReq.TargetPubkey.SerializeCompressed(),	3✔
460	err)	3✔
461	return err	3✔
462
463	// Otherwise, wait for the next channel update. The first update sent
464	// must be the signal to start the PSBT funding in our case since we
465	// specified a PSBT shim. The second update will be the signal that the
466	// channel is now pending.
467	case fundingUpdate := <-channel.updateChan:	11✔
468	log.Tracef("[batchopenchannel] received update: %v",	11✔
469	fundingUpdate)	11✔
470		11✔
471	// Depending on what update we were waiting for the batch	11✔
472	// channel knows what to do with it.	11✔
473	if firstUpdate {	17✔
474	return channel.processPsbtUpdate(fundingUpdate)	6✔
475	}	6✔
476
477	return channel.processPendingUpdate(fundingUpdate)	5✔
478
479	case <-b.cfg.Quit:	×
480	return errShuttingDown	×
481	}
482	}
483
484	// cleanup tries to remove any pending state or UTXO locks in case we had to
485	// abort before finalizing and publishing the funding transaction.
486	func (b *Batcher) cleanup(ctx context.Context) {	4✔
487	// Did we publish a transaction? Then there's nothing to clean up since	4✔
488	// we succeeded.	4✔
489	if b.didPublish {	5✔
490	return	1✔
491	}	1✔
492
493	// Make sure the error message doesn't sound too scary. These might be
494	// logged quite frequently depending on where exactly things were
495	// aborted. We could just not log any cleanup errors though it might be
496	// helpful to debug things if something doesn't go as expected.
497	const errMsgTpl = "Attempted to clean up after failed batch channel " +	3✔
498	"open but could not %s: %v"	3✔
499		3✔
500	// If we failed, we clean up in reverse order. First, let's unlock the	3✔
501	// leased outputs.	3✔
502	for _, lockedUTXO := range b.lockedUTXOs {	5✔
503	rpcOP := &lnrpc.OutPoint{	2✔
504	OutputIndex: lockedUTXO.Outpoint.OutputIndex,	2✔
505	TxidBytes: lockedUTXO.Outpoint.TxidBytes,	2✔
506	TxidStr: lockedUTXO.Outpoint.TxidStr,	2✔
507	}	2✔
508	_, err := b.cfg.WalletKitServer.ReleaseOutput(	2✔
509	ctx, &walletrpc.ReleaseOutputRequest{	2✔
510	Id: lockedUTXO.Id,	2✔
511	Outpoint: rpcOP,	2✔
512	},	2✔
513	)	2✔
514	if err != nil {	2✔
515	log.Debugf(errMsgTpl, "release locked output "+	×
516	lockedUTXO.Outpoint.String(), err)	×
517	}	×
518	}
519
520	// Then go through all channels that ever got into a pending state and
521	// remove the pending channel by abandoning them.
522	for _, channel := range b.channels {	9✔
523	if !channel.isPending {	9✔
524	continue	3✔
525	}
526
527	err := b.cfg.ChannelAbandoner(channel.chanPoint)	3✔
528	if err != nil {	3✔
529	log.Debugf(errMsgTpl, "abandon pending open channel",	×
530	err)	×
531	}	×
532	}
533
534	// And finally clean up the funding shim for each channel that didn't
535	// make it into a pending state.
536	for _, channel := range b.channels {	9✔
537	if channel.isPending {	9✔
538	continue	3✔
539	}
540
541	err := b.cfg.Wallet.CancelFundingIntent(channel.pendingChanID)	3✔
542	if err != nil {	3✔
UNCOV 543	log.Debugf(errMsgTpl, "cancel funding shim", err)	×
UNCOV 544	}	×
545	}
546	}

lightningnetwork / lnd / 13157733617

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous