15561477203

Committed 10 Jun 2025 01:54PM UTC coverage: 58.351% (-10.1%) from 68.487%

Build # 15561477203

Build Type

Pull #9356

github

Committed by

web-flow

Commit Message

Merge 6440b25db into c6d6d4c0b

Pull Request Pull Request #9356: lnrpc: add incoming/outgoing channel ids filter to forwarding history request

Run Details

33 of 36 new or added lines in 2 files covered. (91.67%)

28366 existing lines in 455 files now uncovered.

97715 of 167461 relevant lines covered (58.35%)

1.81 hits per line

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

75.08

/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 {	3✔
49	psbtUpdate := u.GetPsbtFund()	3✔
50	if psbtUpdate == nil {	3✔
51	return fmt.Errorf("got unexpected channel update %v", u.Update)	×
52	}	×
53
54	if psbtUpdate.FundingAmount != int64(c.fundingReq.LocalFundingAmt) {	3✔
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	3✔
61		3✔
62	return nil	3✔
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 {	3✔
68	pendingUpd := u.GetChanPending()	3✔
69	if pendingUpd == nil {	3✔
70	return fmt.Errorf("got unexpected channel update %v", u.Update)	×
71	}	×
72
73	hash, err := chainhash.NewHash(pendingUpd.Txid)	3✔
74	if err != nil {	3✔
75	return fmt.Errorf("could not parse outpoint TX hash: %w", err)	×
76	}	×
77
78	c.chanPoint = &wire.OutPoint{	3✔
79	Index: pendingUpd.OutputIndex,	3✔
80	Hash: *hash,	3✔
81	}	3✔
82	c.isPending = true	3✔
83		3✔
84	return nil	3✔
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 {	3✔
190	return &Batcher{	3✔
191	cfg: cfg,	3✔
192	}	3✔
193	}	3✔
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) {	3✔
200		3✔
201	label, err := labels.ValidateAPI(req.Label)	3✔
202	if err != nil {	3✔
203	return nil, err	×
204	}	×
205
206	// Parse and validate each individual channel.
207	b.channels = make([]*batchChannel, 0, len(req.Channels))	3✔
208	for idx, rpcChannel := range req.Channels {	6✔
209	// If the user specifies a channel ID, it must be exactly 32	3✔
210	// bytes long.	3✔
211	if len(rpcChannel.PendingChanId) > 0 &&	3✔
212	len(rpcChannel.PendingChanId) != 32 {	3✔
213		×
214	return nil, fmt.Errorf("invalid temp chan ID %x",	×
215	rpcChannel.PendingChanId)	×
216	}	×
217
218	var pendingChanID [32]byte	3✔
219	if len(rpcChannel.PendingChanId) == 32 {	3✔
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 {	3✔
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{	3✔
235	SatPerVbyte: uint64(req.SatPerVbyte),	3✔
236	TargetConf: req.TargetConf,	3✔
237	MinConfs: req.MinConfs,	3✔
238	SpendUnconfirmed: req.SpendUnconfirmed,	3✔
239	NodePubkey: rpcChannel.NodePubkey,	3✔
240	LocalFundingAmount: rpcChannel.LocalFundingAmount,	3✔
241	PushSat: rpcChannel.PushSat,	3✔
242	Private: rpcChannel.Private,	3✔
243	MinHtlcMsat: rpcChannel.MinHtlcMsat,	3✔
244	RemoteCsvDelay: rpcChannel.RemoteCsvDelay,	3✔
245	CloseAddress: rpcChannel.CloseAddress,	3✔
246	RemoteMaxValueInFlightMsat: rpcChannel.RemoteMaxValueInFlightMsat,	3✔
247	RemoteMaxHtlcs: rpcChannel.RemoteMaxHtlcs,	3✔
248	MaxLocalCsv: rpcChannel.MaxLocalCsv,	3✔
249	CommitmentType: rpcChannel.CommitmentType,	3✔
250	ZeroConf: rpcChannel.ZeroConf,	3✔
251	ScidAlias: rpcChannel.ScidAlias,	3✔
252	BaseFee: rpcChannel.BaseFee,	3✔
253	FeeRate: rpcChannel.FeeRate,	3✔
254	UseBaseFee: rpcChannel.UseBaseFee,	3✔
255	UseFeeRate: rpcChannel.UseFeeRate,	3✔
256	RemoteChanReserveSat: rpcChannel.RemoteChanReserveSat,	3✔
257	Memo: rpcChannel.Memo,	3✔
258	FundingShim: &lnrpc.FundingShim{	3✔
259	Shim: &lnrpc.FundingShim_PsbtShim{	3✔
260	PsbtShim: &lnrpc.PsbtShim{	3✔
261	PendingChanId: pendingChanID[:],	3✔
262	NoPublish: true,	3✔
263	},	3✔
264	},	3✔
265	},	3✔
266	})	3✔
267	if err != nil {	3✔
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	3✔
275	fundingReq.ChanFunder = chanfunding.NewPsbtAssembler(	3✔
276	btcutil.Amount(rpcChannel.LocalFundingAmount), nil,	3✔
277	b.cfg.NetParams, false,	3✔
278	)	3✔
279		3✔
280	b.channels = append(b.channels, &batchChannel{	3✔
281	pendingChanID: pendingChanID,	3✔
282	fundingReq: fundingReq,	3✔
283	})	3✔
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)	3✔
291		3✔
292	// Now that we know the user input is sane, we need to kick off the	3✔
293	// channel funding negotiation with the peers. Because we specified a	3✔
294	// PSBT assembler, we'll get a special response in the channel once the	3✔
295	// funding output script is known (which we need to craft the TX).	3✔
296	eg := &errgroup.Group{}	3✔
297	for _, channel := range b.channels {	6✔
298	channel.updateChan, channel.errChan = b.cfg.ChannelOpener(	3✔
299	channel.fundingReq,	3✔
300	)	3✔
301		3✔
302	// Launch a goroutine that waits for the initial response on	3✔
303	// either the update or error chan.	3✔
304	channel := channel	3✔
305	eg.Go(func() error {	6✔
306	return b.waitForUpdate(channel, true)	3✔
307	})	3✔
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 {	6✔
313	return nil, fmt.Errorf("error batch opening channel, initial "+	3✔
314	"negotiation failed: %v", err)	3✔
315	}	3✔
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 {	6✔
323	txTemplate.Outputs[channel.fundingAddr] = uint64(	3✔
324	channel.fundingReq.LocalFundingAmt,	3✔
325	)	3✔
326	}	3✔
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 {	6✔
373	err = b.cfg.Wallet.PsbtFundingVerify(	3✔
374	channel.pendingChanID, unsignedPacket, false,	3✔
375	)	3✔
376	if err != nil {	3✔
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 {	6✔
403	err = b.cfg.Wallet.PsbtFundingFinalize(	3✔
404	channel.pendingChanID, nil, finalTx,	3✔
405	)	3✔
406	if err != nil {	3✔
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 {	6✔
416	// Launch another goroutine that waits for the channel pending	3✔
417	// response on the update chan.	3✔
418	channel := channel	3✔
419	eg.Go(func() error {	6✔
420	return b.waitForUpdate(channel, false)	3✔
421	})	3✔
422	}
423
424	// Wait for all updates and make sure we're still good to proceed.
425	if err := eg.Wait(); err != nil {	3✔
UNCOV 426	return nil, fmt.Errorf("error batch opening channel, final "+	×
UNCOV 427	"negotiation failed: %v", err)	×
UNCOV 428	}	×
429
430	// Great, we're now finally ready to publish the transaction.
431	err = b.cfg.Wallet.PublishTransaction(finalTx, label)	3✔
432	if err != nil {	3✔
UNCOV 433	return nil, fmt.Errorf("error publishing final batch "+	×
UNCOV 434	"transaction: %v", err)	×
UNCOV 435	}	×
436	b.didPublish = true	3✔
437		3✔
438	rpcPoints := make([]*lnrpc.PendingUpdate, len(b.channels))	3✔
439	for idx, channel := range b.channels {	6✔
440	rpcPoints[idx] = &lnrpc.PendingUpdate{	3✔
441	Txid: channel.chanPoint.Hash.CloneBytes(),	3✔
442	OutputIndex: channel.chanPoint.Index,	3✔
443	}	3✔
444	}	3✔
445
446	return rpcPoints, nil	3✔
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 {	3✔
455	select {	3✔
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:	3✔
468	log.Tracef("[batchopenchannel] received update: %v",	3✔
469	fundingUpdate)	3✔
470		3✔
471	// Depending on what update we were waiting for the batch	3✔
472	// channel knows what to do with it.	3✔
473	if firstUpdate {	6✔
474	return channel.processPsbtUpdate(fundingUpdate)	3✔
475	}	3✔
476
477	return channel.processPendingUpdate(fundingUpdate)	3✔
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) {	3✔
487	// Did we publish a transaction? Then there's nothing to clean up since	3✔
488	// we succeeded.	3✔
489	if b.didPublish {	6✔
490	return	3✔
491	}	3✔
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 {	3✔
UNCOV 503	rpcOP := &lnrpc.OutPoint{	×
UNCOV 504	OutputIndex: lockedUTXO.Outpoint.OutputIndex,	×
UNCOV 505	TxidBytes: lockedUTXO.Outpoint.TxidBytes,	×
UNCOV 506	TxidStr: lockedUTXO.Outpoint.TxidStr,	×
UNCOV 507	}	×
UNCOV 508	_, err := b.cfg.WalletKitServer.ReleaseOutput(	×
UNCOV 509	ctx, &walletrpc.ReleaseOutputRequest{	×
UNCOV 510	Id: lockedUTXO.Id,	×
UNCOV 511	Outpoint: rpcOP,	×
UNCOV 512	},	×
UNCOV 513	)	×
UNCOV 514	if err != nil {	×
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 {	6✔
523	if !channel.isPending {	6✔
524	continue	3✔
525	}
526
UNCOV 527	err := b.cfg.ChannelAbandoner(channel.chanPoint)	×
UNCOV 528	if err != nil {	×
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 {	6✔
537	if channel.isPending {	3✔
UNCOV 538	continue	×
539	}
540
541	err := b.cfg.Wallet.CancelFundingIntent(channel.pendingChanID)	3✔
542	if err != nil {	6✔
543	log.Debugf(errMsgTpl, "cancel funding shim", err)	3✔
544	}	3✔
545	}
546	}

lightningnetwork / lnd / 15561477203

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