12231552240

Committed 09 Dec 2024 08:17AM UTC coverage: 58.955% (+0.02%) from 58.933%

Build # 12231552240

Build Type

Pull #9242

github

Committed by

aakselrod

Commit Message

go.mod: update btcwallet to latest to eliminate waddrmgr deadlock

Pull Request Pull Request #9242: Reapply #8644

Run Details

24 of 40 new or added lines in 3 files covered. (60.0%)

89 existing lines in 18 files now uncovered.

133525 of 226485 relevant lines covered (58.96%)

19398.62 hits per line

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

62.5

/lnrpc/wtclientrpc/wtclient.go

package wtclientrpc

import (
        "bytes"
        "context"
        "encoding/binary"
        "errors"
        "fmt"
        "net"
        "sort"
        "strconv"

        "github.com/btcsuite/btcd/btcec/v2"
        "github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
        "github.com/lightningnetwork/lnd/lncfg"
        "github.com/lightningnetwork/lnd/lnrpc"
        "github.com/lightningnetwork/lnd/lnwire"
        "github.com/lightningnetwork/lnd/watchtower"
        "github.com/lightningnetwork/lnd/watchtower/blob"
        "github.com/lightningnetwork/lnd/watchtower/wtclient"
        "github.com/lightningnetwork/lnd/watchtower/wtdb"
        "google.golang.org/grpc"
        "gopkg.in/macaroon-bakery.v2/bakery"
)

const (
        // subServerName is the name of the sub rpc server. We'll use this name
        // to register ourselves, and we also require that the main
        // SubServerConfigDispatcher instance recognizes it as the name of our
        // RPC service.
        subServerName = "WatchtowerClientRPC"
)

var (
        // macPermissions maps RPC calls to the permissions they require.
        //
        // TODO(wilmer): create tower macaroon?
        macPermissions = map[string][]bakery.Op{
                "/wtclientrpc.WatchtowerClient/AddTower": {{
                        Entity: "offchain",
                        Action: "write",
                }},
                "/wtclientrpc.WatchtowerClient/RemoveTower": {{
                        Entity: "offchain",
                        Action: "write",
                }},
                "/wtclientrpc.WatchtowerClient/DeactivateTower": {{
                        Entity: "offchain",
                        Action: "write",
                }},
                "/wtclientrpc.WatchtowerClient/TerminateSession": {{
                        Entity: "offchain",
                        Action: "write",
                }},
                "/wtclientrpc.WatchtowerClient/ListTowers": {{
                        Entity: "offchain",
                        Action: "read",
                }},
                "/wtclientrpc.WatchtowerClient/GetTowerInfo": {{
                        Entity: "offchain",
                        Action: "read",
                }},
                "/wtclientrpc.WatchtowerClient/Stats": {{
                        Entity: "offchain",
                        Action: "read",
                }},
                "/wtclientrpc.WatchtowerClient/Policy": {{
                        Entity: "offchain",
                        Action: "read",
                }},
        }

        // ErrWtclientNotActive signals that RPC calls cannot be processed
        // because the watchtower client is not active.
        ErrWtclientNotActive = errors.New("watchtower client not active")
)

// ServerShell is a shell struct holding a reference to the actual sub-server.
// It is used to register the gRPC sub-server with the root server before we
// have the necessary dependencies to populate the actual sub-server.
type ServerShell struct {
        WatchtowerClientServer
}

// WatchtowerClient is the RPC server we'll use to interact with the backing
// active watchtower client.
//
// TODO(wilmer): better name?
type WatchtowerClient struct {
        // Required by the grpc-gateway/v2 library for forward compatibility.
        UnimplementedWatchtowerClientServer

        cfg Config
}

// A compile time check to ensure that WatchtowerClient fully implements the
// WatchtowerClientWatchtowerClient gRPC service.
var _ WatchtowerClientServer = (*WatchtowerClient)(nil)

// New returns a new instance of the wtclientrpc WatchtowerClient sub-server.
// We also return the set of permissions for the macaroons that we may create
// within this method. If the macaroons we need aren't found in the filepath,
// then we'll create them on start up. If we're unable to locate, or create the
// macaroons we need, then we'll return with an error.
func New(cfg *Config) (*WatchtowerClient, lnrpc.MacaroonPerms, error) {
        return &WatchtowerClient{cfg: *cfg}, macPermissions, nil
}

// Start launches any helper goroutines required for the WatchtowerClient to
// function.
//
// NOTE: This is part of the lnrpc.SubWatchtowerClient interface.
func (c *WatchtowerClient) Start() error {
        return nil
}

// Stop signals any active goroutines for a graceful closure.
//
// NOTE: This is part of the lnrpc.SubServer interface.
func (c *WatchtowerClient) Stop() error {
        return nil
}

// Name returns a unique string representation of the sub-server. This can be
// used to identify the sub-server and also de-duplicate them.
//
// NOTE: This is part of the lnrpc.SubServer interface.
func (c *WatchtowerClient) Name() string {
        return subServerName
}

// RegisterWithRootServer will be called by the root gRPC server to direct a sub
// RPC server to register itself with the main gRPC root server. Until this is
// called, each sub-server won't be able to have requests routed towards it.
//
// NOTE: This is part of the lnrpc.GrpcHandler interface.
func (r *ServerShell) RegisterWithRootServer(grpcServer *grpc.Server) error {
        // We make sure that we register it with the main gRPC server to ensure
        // all our methods are routed properly.
        RegisterWatchtowerClientServer(grpcServer, r)

        return nil
}

// RegisterWithRestServer will be called by the root REST mux to direct a sub
// RPC server to register itself with the main REST mux server. Until this is
// called, each sub-server won't be able to have requests routed towards it.
//
// NOTE: This is part of the lnrpc.GrpcHandler interface.
func (r *ServerShell) RegisterWithRestServer(ctx context.Context,
        mux *runtime.ServeMux, dest string, opts []grpc.DialOption) error {

        // We make sure that we register it with the main REST server to ensure
        // all our methods are routed properly.
        err := RegisterWatchtowerClientHandlerFromEndpoint(ctx, mux, dest, opts)
        if err != nil {
                return err
        }

        return nil
}

// CreateSubServer populates the subserver's dependencies using the passed
// SubServerConfigDispatcher. This method should fully initialize the
// sub-server instance, making it ready for action. It returns the macaroon
// permissions that the sub-server wishes to pass on to the root server for all
// methods routed towards it.
//
// NOTE: This is part of the lnrpc.GrpcHandler interface.
func (r *ServerShell) CreateSubServer(configRegistry lnrpc.SubServerConfigDispatcher) (
        lnrpc.SubServer, lnrpc.MacaroonPerms, error) {

        subServer, macPermissions, err := createNewSubServer(configRegistry)
        if err != nil {
                return nil, nil, err
        }

        r.WatchtowerClientServer = subServer
        return subServer, macPermissions, nil
}

// isActive returns nil if the watchtower client is initialized so that we can
// process RPC requests.
func (c *WatchtowerClient) isActive() error {
        if c.cfg.Active {
                return nil
        }
        return ErrWtclientNotActive
}

// AddTower adds a new watchtower reachable at the given address and considers
// it for new sessions. If the watchtower already exists, then any new addresses
// included will be considered when dialing it for session negotiations and
// backups.
func (c *WatchtowerClient) AddTower(ctx context.Context,
        req *AddTowerRequest) (*AddTowerResponse, error) {

        if err := c.isActive(); err != nil {
                return nil, err
        }

        pubKey, err := btcec.ParsePubKey(req.Pubkey)
        if err != nil {
                return nil, err
        }
        addr, err := lncfg.ParseAddressString(
                req.Address, strconv.Itoa(watchtower.DefaultPeerPort),
                c.cfg.Resolver,
        )
        if err != nil {
                return nil, fmt.Errorf("invalid address %v: %w", req.Address,
                        err)
        }

        towerAddr := &lnwire.NetAddress{
                IdentityKey: pubKey,
                Address:     addr,
        }

        if err := c.cfg.ClientMgr.AddTower(towerAddr); err != nil {
                return nil, err
        }

        return &AddTowerResponse{}, nil
}

// RemoveTower removes a watchtower from being considered for future session
// negotiations and from being used for any subsequent backups until it's added
// again. If an address is provided, then this RPC only serves as a way of
// removing the address from the watchtower instead.
func (c *WatchtowerClient) RemoveTower(ctx context.Context,
        req *RemoveTowerRequest) (*RemoveTowerResponse, error) {

        if err := c.isActive(); err != nil {
                return nil, err
        }

        pubKey, err := btcec.ParsePubKey(req.Pubkey)
        if err != nil {
                return nil, err
        }

        var addr net.Addr
        if req.Address != "" {
                addr, err = lncfg.ParseAddressString(
                        req.Address, strconv.Itoa(watchtower.DefaultPeerPort),
                        c.cfg.Resolver,
                )
                if err != nil {
                        return nil, fmt.Errorf("unable to parse tower "+
                                "address %v: %v", req.Address, err)
                }
        }

        err = c.cfg.ClientMgr.RemoveTower(pubKey, addr)
        if err != nil {
                return nil, err
        }

        return &RemoveTowerResponse{}, nil
}

// DeactivateTower sets the given tower's status to inactive so that it is not
// considered for session negotiation. Its sessions will also not be used while
// the tower is inactive.
func (c *WatchtowerClient) DeactivateTower(_ context.Context,
        req *DeactivateTowerRequest) (*DeactivateTowerResponse, error) {

        if err := c.isActive(); err != nil {
                return nil, err
        }

        pubKey, err := btcec.ParsePubKey(req.Pubkey)
        if err != nil {
                return nil, err
        }

        err = c.cfg.ClientMgr.DeactivateTower(pubKey)
        if err != nil {
                return nil, err
        }

        return &DeactivateTowerResponse{
                Status: fmt.Sprintf("Successful deactivation of tower: %x",
                        req.Pubkey),
        }, nil
}

// TerminateSession terminates the given session and marks it as terminal so
// that it is never used again.
func (c *WatchtowerClient) TerminateSession(_ context.Context,
        req *TerminateSessionRequest) (*TerminateSessionResponse, error) {

        if err := c.isActive(); err != nil {
                return nil, err
        }

        pubKey, err := btcec.ParsePubKey(req.SessionId)
        if err != nil {
                return nil, err
        }

        sessionID := wtdb.NewSessionIDFromPubKey(pubKey)

        err = c.cfg.ClientMgr.TerminateSession(sessionID)
        if err != nil {
                return nil, err
        }

        return &TerminateSessionResponse{
                Status: fmt.Sprintf("Successful termination of session: %s",
                        sessionID),
        }, nil
}

// ListTowers returns the list of watchtowers registered with the client.
func (c *WatchtowerClient) ListTowers(ctx context.Context,
        req *ListTowersRequest) (*ListTowersResponse, error) {

        if err := c.isActive(); err != nil {
                return nil, err
        }

        opts, ackCounts, committedUpdateCounts := constructFunctionalOptions(
                req.IncludeSessions, req.ExcludeExhaustedSessions,
        )

        towersPerBlobType, err := c.cfg.ClientMgr.RegisteredTowers(opts...)
        if err != nil {
                return nil, err
        }

        // Collect all the legacy client towers. If it has any of the same
        // towers that the anchors client has, then just add the session info
        // for the legacy client to the existing tower.
        rpcTowers := make(map[wtdb.TowerID]*Tower)
        for blobType, towers := range towersPerBlobType {
                policyType, err := blobTypeToPolicyType(blobType)
                if err != nil {
                        return nil, err
                }

                for _, tower := range towers {
                        rpcTower := marshallTower(
                                tower, policyType, req.IncludeSessions,
                                ackCounts, committedUpdateCounts,
                        )

                        t, ok := rpcTowers[tower.ID]
                        if !ok {
                                rpcTowers[tower.ID] = rpcTower
                                continue
                        }

                        t.SessionInfo = append(
                                t.SessionInfo, rpcTower.SessionInfo...,
                        )
                        t.Sessions = append(
                                t.Sessions, rpcTower.Sessions...,
                        )
                }
        }

        towers := make([]*Tower, 0, len(rpcTowers))
        for _, tower := range rpcTowers {
                towers = append(towers, tower)
        }

        return &ListTowersResponse{Towers: towers}, nil
}

// GetTowerInfo retrieves information for a registered watchtower.
func (c *WatchtowerClient) GetTowerInfo(ctx context.Context,
        req *GetTowerInfoRequest) (*Tower, error) {

        if err := c.isActive(); err != nil {
                return nil, err
        }

        pubKey, err := btcec.ParsePubKey(req.Pubkey)
        if err != nil {
                return nil, err
        }

        opts, ackCounts, committedUpdateCounts := constructFunctionalOptions(
                req.IncludeSessions, req.ExcludeExhaustedSessions,
        )

        towersPerBlobType, err := c.cfg.ClientMgr.LookupTower(pubKey, opts...)
        if err != nil {
                return nil, err
        }

        var resTower *Tower
        for blobType, tower := range towersPerBlobType {
                policyType, err := blobTypeToPolicyType(blobType)
                if err != nil {
                        return nil, err
                }

                rpcTower := marshallTower(
                        tower, policyType, req.IncludeSessions,
                        ackCounts, committedUpdateCounts,
                )

                if resTower == nil {
                        resTower = rpcTower
                        continue
                }

                if !bytes.Equal(rpcTower.Pubkey, resTower.Pubkey) {
                        return nil, fmt.Errorf("tower clients returned " +
                                "inconsistent results for the given tower")
                }

                resTower.SessionInfo = append(
                        resTower.SessionInfo, rpcTower.SessionInfo...,
                )
                resTower.Sessions = append(
                        resTower.Sessions, rpcTower.Sessions...,
                )
        }

        return resTower, nil
}

// constructFunctionalOptions is a helper function that constructs a list of
// functional options to be used when fetching a tower from the DB. It also
// returns a map of acked-update counts and one for un-acked-update counts that
// will be populated once the db call has been made.
func constructFunctionalOptions(includeSessions,
        excludeExhaustedSessions bool) ([]wtdb.ClientSessionListOption,
        map[wtdb.SessionID]uint16, map[wtdb.SessionID]uint16) {

        var (
                opts                  []wtdb.ClientSessionListOption
                committedUpdateCounts = make(map[wtdb.SessionID]uint16)
                ackCounts             = make(map[wtdb.SessionID]uint16)
        )
        if !includeSessions {
                return opts, ackCounts, committedUpdateCounts
        }

        perNumRogueUpdates := func(s *wtdb.ClientSession, numUpdates uint16) {
                ackCounts[s.ID] += numUpdates
        }

        perNumAckedUpdates := func(s *wtdb.ClientSession, id lnwire.ChannelID,
                numUpdates uint16) {

                ackCounts[s.ID] += numUpdates
        }

        perCommittedUpdate := func(s *wtdb.ClientSession,
                u *wtdb.CommittedUpdate) {

                committedUpdateCounts[s.ID]++
        }

        opts = []wtdb.ClientSessionListOption{
                wtdb.WithPerNumAckedUpdates(perNumAckedUpdates),
                wtdb.WithPerCommittedUpdate(perCommittedUpdate),
                wtdb.WithPerRogueUpdateCount(perNumRogueUpdates),
        }

        if excludeExhaustedSessions {
                opts = append(opts, wtdb.WithPostEvalFilterFn(
                        wtclient.ExhaustedSessionFilter(),
                ))
        }

        return opts, ackCounts, committedUpdateCounts
}

// Stats returns the in-memory statistics of the client since startup.
func (c *WatchtowerClient) Stats(_ context.Context,
        _ *StatsRequest) (*StatsResponse, error) {

        if err := c.isActive(); err != nil {
                return nil, err
        }

        stats := c.cfg.ClientMgr.Stats()

        return &StatsResponse{
                NumBackups:           uint32(stats.NumTasksAccepted),
                NumFailedBackups:     uint32(stats.NumTasksIneligible),
                NumPendingBackups:    uint32(stats.NumTasksPending),
                NumSessionsAcquired:  uint32(stats.NumSessionsAcquired),
                NumSessionsExhausted: uint32(stats.NumSessionsExhausted),
        }, nil
}

// Policy returns the active watchtower client policy configuration.
func (c *WatchtowerClient) Policy(ctx context.Context,
        req *PolicyRequest) (*PolicyResponse, error) {

        if err := c.isActive(); err != nil {
                return nil, err
        }

        blobType, err := policyTypeToBlobType(req.PolicyType)
        if err != nil {
                return nil, err
        }

        policy, err := c.cfg.ClientMgr.Policy(blobType)
        if err != nil {
                return nil, err
        }

        return &PolicyResponse{
                MaxUpdates:       uint32(policy.MaxUpdates),
                SweepSatPerVbyte: uint32(policy.SweepFeeRate.FeePerVByte()),

                // Deprecated field.
                SweepSatPerByte: uint32(policy.SweepFeeRate.FeePerVByte()),
        }, nil
}

// marshallTower converts a client registered watchtower into its corresponding
// RPC type.
func marshallTower(tower *wtclient.RegisteredTower, policyType PolicyType,
        includeSessions bool, ackCounts map[wtdb.SessionID]uint16,
        pendingCounts map[wtdb.SessionID]uint16) *Tower {

        rpcAddrs := make([]string, 0, len(tower.Addresses))
        for _, addr := range tower.Addresses {
                rpcAddrs = append(rpcAddrs, addr.String())
        }

        var rpcSessions []*TowerSession
        if includeSessions {
                // To ensure that the output order is deterministic for a given
                // set of sessions, we put the sessions into a slice and order
                // them based on session ID.
                sessions := make([]*wtdb.ClientSession, 0, len(tower.Sessions))
                for _, session := range tower.Sessions {
                        sessions = append(sessions, session)
                }

                sort.Slice(sessions, func(i, j int) bool {
                        id1 := sessions[i].ID
                        id2 := sessions[j].ID

                        return binary.BigEndian.Uint64(id1[:]) <
                                binary.BigEndian.Uint64(id2[:])
                })

                rpcSessions = make([]*TowerSession, 0, len(tower.Sessions))
                for _, session := range sessions {
                        satPerVByte := session.Policy.SweepFeeRate.FeePerVByte()
                        rpcSessions = append(rpcSessions, &TowerSession{
                                Id:                session.ID[:],
                                NumBackups:        uint32(ackCounts[session.ID]),
                                NumPendingBackups: uint32(pendingCounts[session.ID]),
                                MaxBackups:        uint32(session.Policy.MaxUpdates),
                                SweepSatPerVbyte:  uint32(satPerVByte),

                                // Deprecated field.
                                SweepSatPerByte: uint32(satPerVByte),
                        })
                }
        }

        rpcTower := &Tower{
                Pubkey:    tower.IdentityKey.SerializeCompressed(),
                Addresses: rpcAddrs,
                SessionInfo: []*TowerSessionInfo{{
                        PolicyType:             policyType,
                        ActiveSessionCandidate: tower.ActiveSessionCandidate,
                        NumSessions:            uint32(len(tower.Sessions)),
                        Sessions:               rpcSessions,
                }},
                // The below fields are populated for backwards compatibility
                // but will be removed in a future commit when the proto fields
                // are removed.
                ActiveSessionCandidate: tower.ActiveSessionCandidate,
                NumSessions:            uint32(len(tower.Sessions)),
                Sessions:               rpcSessions,
        }

        return rpcTower
}

func blobTypeToPolicyType(t blob.Type) (PolicyType, error) {
        switch t {
        case blob.TypeAltruistTaprootCommit:
                return PolicyType_TAPROOT, nil

        case blob.TypeAltruistAnchorCommit:
                return PolicyType_ANCHOR, nil

        case blob.TypeAltruistCommit:
                return PolicyType_LEGACY, nil

        default:
                return 0, fmt.Errorf("unknown blob type: %s", t)
        }
}

func policyTypeToBlobType(t PolicyType) (blob.Type, error) {
        switch t {
        case PolicyType_TAPROOT:
                return blob.TypeAltruistTaprootCommit, nil

        case PolicyType_ANCHOR:
                return blob.TypeAltruistAnchorCommit, nil

        case PolicyType_LEGACY:
                return blob.TypeAltruistCommit, nil

        default:
                return 0, fmt.Errorf("unknown policy type: %s", t)
        }
}

1	package wtclientrpc
2
3	import (
4	"bytes"
5	"context"
6	"encoding/binary"
7	"errors"
8	"fmt"
9	"net"
10	"sort"
11	"strconv"
12
13	"github.com/btcsuite/btcd/btcec/v2"
14	"github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
15	"github.com/lightningnetwork/lnd/lncfg"
16	"github.com/lightningnetwork/lnd/lnrpc"
17	"github.com/lightningnetwork/lnd/lnwire"
18	"github.com/lightningnetwork/lnd/watchtower"
19	"github.com/lightningnetwork/lnd/watchtower/blob"
20	"github.com/lightningnetwork/lnd/watchtower/wtclient"
21	"github.com/lightningnetwork/lnd/watchtower/wtdb"
22	"google.golang.org/grpc"
23	"gopkg.in/macaroon-bakery.v2/bakery"
24	)
25
26	const (
27	// subServerName is the name of the sub rpc server. We'll use this name
28	// to register ourselves, and we also require that the main
29	// SubServerConfigDispatcher instance recognizes it as the name of our
30	// RPC service.
31	subServerName = "WatchtowerClientRPC"
32	)
33
34	var (
35	// macPermissions maps RPC calls to the permissions they require.
36	//
37	// TODO(wilmer): create tower macaroon?
38	macPermissions = map[string][]bakery.Op{
39	"/wtclientrpc.WatchtowerClient/AddTower": {{
40	Entity: "offchain",
41	Action: "write",
42	}},
43	"/wtclientrpc.WatchtowerClient/RemoveTower": {{
44	Entity: "offchain",
45	Action: "write",
46	}},
47	"/wtclientrpc.WatchtowerClient/DeactivateTower": {{
48	Entity: "offchain",
49	Action: "write",
50	}},
51	"/wtclientrpc.WatchtowerClient/TerminateSession": {{
52	Entity: "offchain",
53	Action: "write",
54	}},
55	"/wtclientrpc.WatchtowerClient/ListTowers": {{
56	Entity: "offchain",
57	Action: "read",
58	}},
59	"/wtclientrpc.WatchtowerClient/GetTowerInfo": {{
60	Entity: "offchain",
61	Action: "read",
62	}},
63	"/wtclientrpc.WatchtowerClient/Stats": {{
64	Entity: "offchain",
65	Action: "read",
66	}},
67	"/wtclientrpc.WatchtowerClient/Policy": {{
68	Entity: "offchain",
69	Action: "read",
70	}},
71	}
72
73	// ErrWtclientNotActive signals that RPC calls cannot be processed
74	// because the watchtower client is not active.
75	ErrWtclientNotActive = errors.New("watchtower client not active")
76	)
77
78	// ServerShell is a shell struct holding a reference to the actual sub-server.
79	// It is used to register the gRPC sub-server with the root server before we
80	// have the necessary dependencies to populate the actual sub-server.
81	type ServerShell struct {
82	WatchtowerClientServer
83	}
84
85	// WatchtowerClient is the RPC server we'll use to interact with the backing
86	// active watchtower client.
87	//
88	// TODO(wilmer): better name?
89	type WatchtowerClient struct {
90	// Required by the grpc-gateway/v2 library for forward compatibility.
91	UnimplementedWatchtowerClientServer
92
93	cfg Config
94	}
95
96	// A compile time check to ensure that WatchtowerClient fully implements the
97	// WatchtowerClientWatchtowerClient gRPC service.
98	var _ WatchtowerClientServer = (*WatchtowerClient)(nil)
99
100	// New returns a new instance of the wtclientrpc WatchtowerClient sub-server.
101	// We also return the set of permissions for the macaroons that we may create
102	// within this method. If the macaroons we need aren't found in the filepath,
103	// then we'll create them on start up. If we're unable to locate, or create the
104	// macaroons we need, then we'll return with an error.
105	func New(cfg Config) (WatchtowerClient, lnrpc.MacaroonPerms, error) {	4✔
106	return &WatchtowerClient{cfg: *cfg}, macPermissions, nil	4✔
107	}	4✔
108
109	// Start launches any helper goroutines required for the WatchtowerClient to
110	// function.
111	//
112	// NOTE: This is part of the lnrpc.SubWatchtowerClient interface.
113	func (c *WatchtowerClient) Start() error {	4✔
114	return nil	4✔
115	}	4✔
116
117	// Stop signals any active goroutines for a graceful closure.
118	//
119	// NOTE: This is part of the lnrpc.SubServer interface.
120	func (c *WatchtowerClient) Stop() error {	4✔
121	return nil	4✔
122	}	4✔
123
124	// Name returns a unique string representation of the sub-server. This can be
125	// used to identify the sub-server and also de-duplicate them.
126	//
127	// NOTE: This is part of the lnrpc.SubServer interface.
128	func (c *WatchtowerClient) Name() string {	4✔
129	return subServerName	4✔
130	}	4✔
131
132	// RegisterWithRootServer will be called by the root gRPC server to direct a sub
133	// RPC server to register itself with the main gRPC root server. Until this is
134	// called, each sub-server won't be able to have requests routed towards it.
135	//
136	// NOTE: This is part of the lnrpc.GrpcHandler interface.
137	func (r ServerShell) RegisterWithRootServer(grpcServer grpc.Server) error {	4✔
138	// We make sure that we register it with the main gRPC server to ensure	4✔
139	// all our methods are routed properly.	4✔
140	RegisterWatchtowerClientServer(grpcServer, r)	4✔
141		4✔
142	return nil	4✔
143	}	4✔
144
145	// RegisterWithRestServer will be called by the root REST mux to direct a sub
146	// RPC server to register itself with the main REST mux server. Until this is
147	// called, each sub-server won't be able to have requests routed towards it.
148	//
149	// NOTE: This is part of the lnrpc.GrpcHandler interface.
150	func (r *ServerShell) RegisterWithRestServer(ctx context.Context,
151	mux *runtime.ServeMux, dest string, opts []grpc.DialOption) error {	4✔
152		4✔
153	// We make sure that we register it with the main REST server to ensure	4✔
154	// all our methods are routed properly.	4✔
155	err := RegisterWatchtowerClientHandlerFromEndpoint(ctx, mux, dest, opts)	4✔
156	if err != nil {	4✔
157	return err	×
158	}	×
159
160	return nil	4✔
161	}
162
163	// CreateSubServer populates the subserver's dependencies using the passed
164	// SubServerConfigDispatcher. This method should fully initialize the
165	// sub-server instance, making it ready for action. It returns the macaroon
166	// permissions that the sub-server wishes to pass on to the root server for all
167	// methods routed towards it.
168	//
169	// NOTE: This is part of the lnrpc.GrpcHandler interface.
170	func (r *ServerShell) CreateSubServer(configRegistry lnrpc.SubServerConfigDispatcher) (
171	lnrpc.SubServer, lnrpc.MacaroonPerms, error) {	4✔
172		4✔
173	subServer, macPermissions, err := createNewSubServer(configRegistry)	4✔
174	if err != nil {	4✔
175	return nil, nil, err	×
176	}	×
177
178	r.WatchtowerClientServer = subServer	4✔
179	return subServer, macPermissions, nil	4✔
180	}
181
182	// isActive returns nil if the watchtower client is initialized so that we can
183	// process RPC requests.
184	func (c *WatchtowerClient) isActive() error {	4✔
185	if c.cfg.Active {	8✔
186	return nil	4✔
187	}	4✔
188	return ErrWtclientNotActive	×
189	}
190
191	// AddTower adds a new watchtower reachable at the given address and considers
192	// it for new sessions. If the watchtower already exists, then any new addresses
193	// included will be considered when dialing it for session negotiations and
194	// backups.
195	func (c *WatchtowerClient) AddTower(ctx context.Context,
196	req AddTowerRequest) (AddTowerResponse, error) {	4✔
197		4✔
198	if err := c.isActive(); err != nil {	4✔
199	return nil, err	×
200	}	×
201
202	pubKey, err := btcec.ParsePubKey(req.Pubkey)	4✔
203	if err != nil {	4✔
204	return nil, err	×
205	}	×
206	addr, err := lncfg.ParseAddressString(	4✔
207	req.Address, strconv.Itoa(watchtower.DefaultPeerPort),	4✔
208	c.cfg.Resolver,	4✔
209	)	4✔
210	if err != nil {	4✔
211	return nil, fmt.Errorf("invalid address %v: %w", req.Address,	×
212	err)	×
213	}	×
214
215	towerAddr := &lnwire.NetAddress{	4✔
216	IdentityKey: pubKey,	4✔
217	Address: addr,	4✔
218	}	4✔
219		4✔
220	if err := c.cfg.ClientMgr.AddTower(towerAddr); err != nil {	4✔
221	return nil, err	×
222	}	×
223
224	return &AddTowerResponse{}, nil	4✔
225	}
226
227	// RemoveTower removes a watchtower from being considered for future session
228	// negotiations and from being used for any subsequent backups until it's added
229	// again. If an address is provided, then this RPC only serves as a way of
230	// removing the address from the watchtower instead.
231	func (c *WatchtowerClient) RemoveTower(ctx context.Context,
232	req RemoveTowerRequest) (RemoveTowerResponse, error) {	4✔
233		4✔
234	if err := c.isActive(); err != nil {	4✔
235	return nil, err	×
236	}	×
237
238	pubKey, err := btcec.ParsePubKey(req.Pubkey)	4✔
239	if err != nil {	4✔
240	return nil, err	×
241	}	×
242
243	var addr net.Addr	4✔
244	if req.Address != "" {	4✔
245	addr, err = lncfg.ParseAddressString(	×
246	req.Address, strconv.Itoa(watchtower.DefaultPeerPort),	×
247	c.cfg.Resolver,	×
248	)	×
249	if err != nil {	×
250	return nil, fmt.Errorf("unable to parse tower "+	×
251	"address %v: %v", req.Address, err)	×
252	}	×
253	}
254
255	err = c.cfg.ClientMgr.RemoveTower(pubKey, addr)	4✔
256	if err != nil {	4✔
257	return nil, err	×
258	}	×
259
260	return &RemoveTowerResponse{}, nil	4✔
261	}
262
263	// DeactivateTower sets the given tower's status to inactive so that it is not
264	// considered for session negotiation. Its sessions will also not be used while
265	// the tower is inactive.
266	func (c *WatchtowerClient) DeactivateTower(_ context.Context,
267	req DeactivateTowerRequest) (DeactivateTowerResponse, error) {	4✔
268		4✔
269	if err := c.isActive(); err != nil {	4✔
270	return nil, err	×
271	}	×
272
273	pubKey, err := btcec.ParsePubKey(req.Pubkey)	4✔
274	if err != nil {	4✔
275	return nil, err	×
276	}	×
277
278	err = c.cfg.ClientMgr.DeactivateTower(pubKey)	4✔
279	if err != nil {	4✔
280	return nil, err	×
281	}	×
282
283	return &DeactivateTowerResponse{	4✔
284	Status: fmt.Sprintf("Successful deactivation of tower: %x",	4✔
285	req.Pubkey),	4✔
286	}, nil	4✔
287	}
288
289	// TerminateSession terminates the given session and marks it as terminal so
290	// that it is never used again.
291	func (c *WatchtowerClient) TerminateSession(_ context.Context,
292	req TerminateSessionRequest) (TerminateSessionResponse, error) {	4✔
293		4✔
294	if err := c.isActive(); err != nil {	4✔
295	return nil, err	×
296	}	×
297
298	pubKey, err := btcec.ParsePubKey(req.SessionId)	4✔
299	if err != nil {	4✔
300	return nil, err	×
301	}	×
302
303	sessionID := wtdb.NewSessionIDFromPubKey(pubKey)	4✔
304		4✔
305	err = c.cfg.ClientMgr.TerminateSession(sessionID)	4✔
306	if err != nil {	4✔
307	return nil, err	×
308	}	×
309
310	return &TerminateSessionResponse{	4✔
311	Status: fmt.Sprintf("Successful termination of session: %s",	4✔
312	sessionID),	4✔
313	}, nil	4✔
314	}
315
316	// ListTowers returns the list of watchtowers registered with the client.
317	func (c *WatchtowerClient) ListTowers(ctx context.Context,
318	req ListTowersRequest) (ListTowersResponse, error) {	×
319		×
320	if err := c.isActive(); err != nil {	×
321	return nil, err	×
322	}	×
323
324	opts, ackCounts, committedUpdateCounts := constructFunctionalOptions(	×
325	req.IncludeSessions, req.ExcludeExhaustedSessions,	×
326	)	×
327		×
328	towersPerBlobType, err := c.cfg.ClientMgr.RegisteredTowers(opts...)	×
329	if err != nil {	×
330	return nil, err	×
331	}	×
332
333	// Collect all the legacy client towers. If it has any of the same
334	// towers that the anchors client has, then just add the session info
335	// for the legacy client to the existing tower.
336	rpcTowers := make(map[wtdb.TowerID]*Tower)	×
337	for blobType, towers := range towersPerBlobType {	×
338	policyType, err := blobTypeToPolicyType(blobType)	×
339	if err != nil {	×
340	return nil, err	×
341	}	×
342
343	for _, tower := range towers {	×
344	rpcTower := marshallTower(	×
345	tower, policyType, req.IncludeSessions,	×
346	ackCounts, committedUpdateCounts,	×
347	)	×
348		×
349	t, ok := rpcTowers[tower.ID]	×
350	if !ok {	×
351	rpcTowers[tower.ID] = rpcTower	×
352	continue	×
353	}
354
355	t.SessionInfo = append(	×
356	t.SessionInfo, rpcTower.SessionInfo...,	×
357	)	×
358	t.Sessions = append(	×
359	t.Sessions, rpcTower.Sessions...,	×
360	)	×
361	}
362	}
363
364	towers := make([]*Tower, 0, len(rpcTowers))	×
365	for _, tower := range rpcTowers {	×
366	towers = append(towers, tower)	×
367	}	×
368
369	return &ListTowersResponse{Towers: towers}, nil	×
370	}
371
372	// GetTowerInfo retrieves information for a registered watchtower.
373	func (c *WatchtowerClient) GetTowerInfo(ctx context.Context,
374	req GetTowerInfoRequest) (Tower, error) {	4✔
375		4✔
376	if err := c.isActive(); err != nil {	4✔
377	return nil, err	×
378	}	×
379
380	pubKey, err := btcec.ParsePubKey(req.Pubkey)	4✔
381	if err != nil {	4✔
382	return nil, err	×
383	}	×
384
385	opts, ackCounts, committedUpdateCounts := constructFunctionalOptions(	4✔
386	req.IncludeSessions, req.ExcludeExhaustedSessions,	4✔
387	)	4✔
388		4✔
389	towersPerBlobType, err := c.cfg.ClientMgr.LookupTower(pubKey, opts...)	4✔
390	if err != nil {	4✔
391	return nil, err	×
392	}	×
393
394	var resTower *Tower	4✔
395	for blobType, tower := range towersPerBlobType {	8✔
396	policyType, err := blobTypeToPolicyType(blobType)	4✔
397	if err != nil {	4✔
398	return nil, err	×
399	}	×
400
401	rpcTower := marshallTower(	4✔
402	tower, policyType, req.IncludeSessions,	4✔
403	ackCounts, committedUpdateCounts,	4✔
404	)	4✔
405		4✔
406	if resTower == nil {	8✔
407	resTower = rpcTower	4✔
408	continue	4✔
409	}
410
411	if !bytes.Equal(rpcTower.Pubkey, resTower.Pubkey) {	4✔
412	return nil, fmt.Errorf("tower clients returned " +	×
413	"inconsistent results for the given tower")	×
414	}	×
415
416	resTower.SessionInfo = append(	4✔
417	resTower.SessionInfo, rpcTower.SessionInfo...,	4✔
418	)	4✔
419	resTower.Sessions = append(	4✔
420	resTower.Sessions, rpcTower.Sessions...,	4✔
421	)	4✔
422	}
423
424	return resTower, nil	4✔
425	}
426
427	// constructFunctionalOptions is a helper function that constructs a list of
428	// functional options to be used when fetching a tower from the DB. It also
429	// returns a map of acked-update counts and one for un-acked-update counts that
430	// will be populated once the db call has been made.
431	func constructFunctionalOptions(includeSessions,
432	excludeExhaustedSessions bool) ([]wtdb.ClientSessionListOption,
433	map[wtdb.SessionID]uint16, map[wtdb.SessionID]uint16) {	4✔
434		4✔
435	var (	4✔
436	opts []wtdb.ClientSessionListOption	4✔
437	committedUpdateCounts = make(map[wtdb.SessionID]uint16)	4✔
438	ackCounts = make(map[wtdb.SessionID]uint16)	4✔
439	)	4✔
440	if !includeSessions {	8✔
441	return opts, ackCounts, committedUpdateCounts	4✔
442	}	4✔
443
444	perNumRogueUpdates := func(s *wtdb.ClientSession, numUpdates uint16) {	8✔
445	ackCounts[s.ID] += numUpdates	4✔
446	}	4✔
447
448	perNumAckedUpdates := func(s *wtdb.ClientSession, id lnwire.ChannelID,	4✔
449	numUpdates uint16) {	8✔
450		4✔
451	ackCounts[s.ID] += numUpdates	4✔
452	}	4✔
453
454	perCommittedUpdate := func(s *wtdb.ClientSession,	4✔
455	u *wtdb.CommittedUpdate) {	4✔
UNCOV 456		×
UNCOV 457	committedUpdateCounts[s.ID]++	×
UNCOV 458	}	×
459
460	opts = []wtdb.ClientSessionListOption{	4✔
461	wtdb.WithPerNumAckedUpdates(perNumAckedUpdates),	4✔
462	wtdb.WithPerCommittedUpdate(perCommittedUpdate),	4✔
463	wtdb.WithPerRogueUpdateCount(perNumRogueUpdates),	4✔
464	}	4✔
465		4✔
466	if excludeExhaustedSessions {	4✔
467	opts = append(opts, wtdb.WithPostEvalFilterFn(	×
468	wtclient.ExhaustedSessionFilter(),	×
469	))	×
470	}	×
471
472	return opts, ackCounts, committedUpdateCounts	4✔
473	}
474
475	// Stats returns the in-memory statistics of the client since startup.
476	func (c *WatchtowerClient) Stats(_ context.Context,
477	_ StatsRequest) (StatsResponse, error) {	4✔
478		4✔
479	if err := c.isActive(); err != nil {	4✔
480	return nil, err	×
481	}	×
482
483	stats := c.cfg.ClientMgr.Stats()	4✔
484		4✔
485	return &StatsResponse{	4✔
486	NumBackups: uint32(stats.NumTasksAccepted),	4✔
487	NumFailedBackups: uint32(stats.NumTasksIneligible),	4✔
488	NumPendingBackups: uint32(stats.NumTasksPending),	4✔
489	NumSessionsAcquired: uint32(stats.NumSessionsAcquired),	4✔
490	NumSessionsExhausted: uint32(stats.NumSessionsExhausted),	4✔
491	}, nil	4✔
492	}
493
494	// Policy returns the active watchtower client policy configuration.
495	func (c *WatchtowerClient) Policy(ctx context.Context,
496	req PolicyRequest) (PolicyResponse, error) {	×
497		×
498	if err := c.isActive(); err != nil {	×
499	return nil, err	×
500	}	×
501
502	blobType, err := policyTypeToBlobType(req.PolicyType)	×
503	if err != nil {	×
504	return nil, err	×
505	}	×
506
507	policy, err := c.cfg.ClientMgr.Policy(blobType)	×
508	if err != nil {	×
509	return nil, err	×
510	}	×
511
512	return &PolicyResponse{	×
513	MaxUpdates: uint32(policy.MaxUpdates),	×
514	SweepSatPerVbyte: uint32(policy.SweepFeeRate.FeePerVByte()),	×
515		×
516	// Deprecated field.	×
517	SweepSatPerByte: uint32(policy.SweepFeeRate.FeePerVByte()),	×
518	}, nil	×
519	}
520
521	// marshallTower converts a client registered watchtower into its corresponding
522	// RPC type.
523	func marshallTower(tower *wtclient.RegisteredTower, policyType PolicyType,
524	includeSessions bool, ackCounts map[wtdb.SessionID]uint16,
525	pendingCounts map[wtdb.SessionID]uint16) *Tower {	4✔
526		4✔
527	rpcAddrs := make([]string, 0, len(tower.Addresses))	4✔
528	for _, addr := range tower.Addresses {	8✔
529	rpcAddrs = append(rpcAddrs, addr.String())	4✔
530	}	4✔
531
532	var rpcSessions []*TowerSession	4✔
533	if includeSessions {	8✔
534	// To ensure that the output order is deterministic for a given	4✔
535	// set of sessions, we put the sessions into a slice and order	4✔
536	// them based on session ID.	4✔
537	sessions := make([]*wtdb.ClientSession, 0, len(tower.Sessions))	4✔
538	for _, session := range tower.Sessions {	8✔
539	sessions = append(sessions, session)	4✔
540	}	4✔
541
542	sort.Slice(sessions, func(i, j int) bool {	8✔
543	id1 := sessions[i].ID	4✔
544	id2 := sessions[j].ID	4✔
545		4✔
546	return binary.BigEndian.Uint64(id1[:]) <	4✔
547	binary.BigEndian.Uint64(id2[:])	4✔
548	})	4✔
549
550	rpcSessions = make([]*TowerSession, 0, len(tower.Sessions))	4✔
551	for _, session := range sessions {	8✔
552	satPerVByte := session.Policy.SweepFeeRate.FeePerVByte()	4✔
553	rpcSessions = append(rpcSessions, &TowerSession{	4✔
554	Id: session.ID[:],	4✔
555	NumBackups: uint32(ackCounts[session.ID]),	4✔
556	NumPendingBackups: uint32(pendingCounts[session.ID]),	4✔
557	MaxBackups: uint32(session.Policy.MaxUpdates),	4✔
558	SweepSatPerVbyte: uint32(satPerVByte),	4✔
559		4✔
560	// Deprecated field.	4✔
561	SweepSatPerByte: uint32(satPerVByte),	4✔
562	})	4✔
563	}	4✔
564	}
565
566	rpcTower := &Tower{	4✔
567	Pubkey: tower.IdentityKey.SerializeCompressed(),	4✔
568	Addresses: rpcAddrs,	4✔
569	SessionInfo: []*TowerSessionInfo{{	4✔
570	PolicyType: policyType,	4✔
571	ActiveSessionCandidate: tower.ActiveSessionCandidate,	4✔
572	NumSessions: uint32(len(tower.Sessions)),	4✔
573	Sessions: rpcSessions,	4✔
574	}},	4✔
575	// The below fields are populated for backwards compatibility	4✔
576	// but will be removed in a future commit when the proto fields	4✔
577	// are removed.	4✔
578	ActiveSessionCandidate: tower.ActiveSessionCandidate,	4✔
579	NumSessions: uint32(len(tower.Sessions)),	4✔
580	Sessions: rpcSessions,	4✔
581	}	4✔
582		4✔
583	return rpcTower	4✔
584	}
585
586	func blobTypeToPolicyType(t blob.Type) (PolicyType, error) {	4✔
587	switch t {	4✔
588	case blob.TypeAltruistTaprootCommit:	4✔
589	return PolicyType_TAPROOT, nil	4✔
590
591	case blob.TypeAltruistAnchorCommit:	4✔
592	return PolicyType_ANCHOR, nil	4✔
593
594	case blob.TypeAltruistCommit:	4✔
595	return PolicyType_LEGACY, nil	4✔
596
597	default:	×
598	return 0, fmt.Errorf("unknown blob type: %s", t)	×
599	}
600	}
601
602	func policyTypeToBlobType(t PolicyType) (blob.Type, error) {	×
603	switch t {	×
604	case PolicyType_TAPROOT:	×
605	return blob.TypeAltruistTaprootCommit, nil	×
606
607	case PolicyType_ANCHOR:	×
608	return blob.TypeAltruistAnchorCommit, nil	×
609
610	case PolicyType_LEGACY:	×
611	return blob.TypeAltruistCommit, nil	×
612
613	default:	×
614	return 0, fmt.Errorf("unknown policy type: %s", t)	×
615	}
616	}

lightningnetwork / lnd / 12231552240

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