12252204654

Committed 10 Dec 2024 08:24AM UTC coverage: 49.833% (+0.06%) from 49.773%

Build # 12252204654

Build Type

Pull #9343

github

Committed by

ellemouton

Commit Message

fn: rework the ContextGuard and add tests

In this commit, the ContextGuard struct is re-worked such that the
context that its new main WithCtx method provides is cancelled in sync
with a parent context being cancelled or with it's quit channel being
cancelled. Tests are added to assert the behaviour. In order for the
close of the quit channel to be consistent with the cancelling of the
derived context, the quit channel _must_ be contained internal to the
ContextGuard so that callers are only able to close the channel via the
exposed Quit method which will then take care to first cancel any
derived context that depend on the quit channel before returning.

Pull Request Pull Request #9343: fn: expand the ContextGuard and add tests

Run Details

100202 of 201077 relevant lines covered (49.83%)

2.07 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✔
456		×
457	committedUpdateCounts[s.ID]++	×
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 / 12252204654

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