18852986778

Committed 27 Oct 2025 07:10PM UTC coverage: 54.859% (-11.8%) from 66.648%

Build # 18852986778

Build Type

Pull #10265

github

Committed by

web-flow

Commit Message

Merge 45787b3d5 into 9a7b526c0

Pull Request Pull Request #10265: multi: update close logic to handle re-orgs of depth n-1, where n is num confs - add min conf floor

Run Details

529 of 828 new or added lines in 17 files covered. (63.89%)

24026 existing lines in 286 files now uncovered.

110927 of 202205 relevant lines covered (54.86%)

21658.16 hits per line

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

0.0

/lnrpc/devrpc/dev_server.go

//go:build dev
// +build dev

package devrpc

import (
        "context"
        "encoding/hex"
        "fmt"
        "strconv"
        "strings"
        "sync/atomic"
        "time"

        "github.com/btcsuite/btcd/btcutil"
        "github.com/btcsuite/btcd/chaincfg/chainhash"
        "github.com/btcsuite/btcd/wire"
        "github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
        "github.com/lightningnetwork/lnd/fn/v2"
        "github.com/lightningnetwork/lnd/graph/db/models"
        "github.com/lightningnetwork/lnd/lncfg"
        "github.com/lightningnetwork/lnd/lnrpc"
        "github.com/lightningnetwork/lnd/lntypes"
        "github.com/lightningnetwork/lnd/lnwire"
        "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 recognize tt as the name of our
        // RPC service.
        subServerName = "DevRPC"
)

var (
        // macPermissions maps RPC calls to the permissions they require.
        macPermissions = map[string][]bakery.Op{
                "/devrpc.Dev/ImportGraph": {{
                        Entity: "offchain",
                        Action: "write",
                }},
                "/devrpc.Dev/Quiesce": {{
                        Entity: "offchain",
                        Action: "write",
                }},
                "/devrpc.Dev/TriggerSweeper": {{
                        Entity: "onchain",
                        Action: "write",
                }},
        }
)

// 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 {
        DevServer
}

// Server is a sub-server of the main RPC server: the dev RPC. This sub
// RPC server allows developers to set and query LND state that is not possible
// during normal operation.
type Server struct {
        started  int32 // To be used atomically.
        shutdown int32 // To be used atomically.
        quit     chan struct{}

        // Required by the grpc-gateway/v2 library for forward compatibility.
        // Must be after the atomically used variables to not break struct
        // alignment.
        UnimplementedDevServer

        cfg *Config
}

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

// New returns a new instance of the devrpc Dev 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) (*Server, lnrpc.MacaroonPerms, error) {
        // We don't create any new macaroons for this subserver, instead reuse
        // existing onchain/offchain permissions.
        server := &Server{
                quit: make(chan struct{}),
                cfg:  cfg,
        }

        return server, macPermissions, nil
}

// Start launches any helper goroutines required for the Server to function.
//
// NOTE: This is part of the lnrpc.SubServer interface.
func (s *Server) Start() error {
        if atomic.AddInt32(&s.started, 1) != 1 {
                return nil
        }

        return nil
}

// Stop signals any active goroutines for a graceful closure.
//
// NOTE: This is part of the lnrpc.SubServer interface.
func (s *Server) Stop() error {
        if atomic.AddInt32(&s.shutdown, 1) != 1 {
                return nil
        }

        close(s.quit)

        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 (s *Server) 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.
        RegisterDevServer(grpcServer, r)

        log.Debugf("DEV RPC server successfully registered with root the " +
                "gRPC server")

        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 := RegisterDevHandlerFromEndpoint(ctx, mux, dest, opts)
        if err != nil {
                log.Errorf("Could not register DEV REST server with the root "+
                        "REST server: %v", err)
                return err
        }

        log.Debugf("DEV REST server successfully registered with the root " +
                "REST server")
        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.DevServer = subServer
        return subServer, macPermissions, nil
}

func parseOutPoint(s string) (*wire.OutPoint, error) {
        split := strings.Split(s, ":")
        if len(split) != 2 {
                return nil, fmt.Errorf("expecting outpoint to be in format of: " +
                        "txid:index")
        }

        index, err := strconv.ParseInt(split[1], 10, 32)
        if err != nil {
                return nil, fmt.Errorf("unable to decode output index: %w", err)
        }

        txid, err := chainhash.NewHashFromStr(split[0])
        if err != nil {
                return nil, fmt.Errorf("unable to parse hex string: %w", err)
        }

        return &wire.OutPoint{
                Hash:  *txid,
                Index: uint32(index),
        }, nil
}

func parsePubKey(pubKeyStr string) ([33]byte, error) {
        var pubKey [33]byte
        pubKeyBytes, err := hex.DecodeString(pubKeyStr)
        if err != nil || len(pubKeyBytes) != 33 {
                return pubKey, fmt.Errorf("invalid pubkey: %v", pubKeyStr)
        }

        copy(pubKey[:], pubKeyBytes)
        return pubKey, nil
}

// ImportGraph imports a graph dump (without auth proofs).
//
// NOTE: Part of the DevServer interface.
func (s *Server) ImportGraph(ctx context.Context,
        graph *lnrpc.ChannelGraph) (*ImportGraphResponse, error) {

        // Obtain the pointer to the global singleton channel graph.
        graphDB := s.cfg.GraphDB

        var err error
        for _, rpcNode := range graph.Nodes {
                node := &models.Node{
                        HaveNodeAnnouncement: true,
                        LastUpdate: time.Unix(
                                int64(rpcNode.LastUpdate), 0,
                        ),
                        Alias: rpcNode.Alias,
                }

                node.PubKeyBytes, err = parsePubKey(rpcNode.PubKey)
                if err != nil {
                        return nil, err
                }

                featureBits := make([]lnwire.FeatureBit, 0, len(rpcNode.Features))
                featureNames := make(map[lnwire.FeatureBit]string)

                for featureBit, feature := range rpcNode.Features {
                        featureBits = append(
                                featureBits, lnwire.FeatureBit(featureBit),
                        )

                        featureNames[lnwire.FeatureBit(featureBit)] = feature.Name
                }

                featureVector := lnwire.NewRawFeatureVector(featureBits...)
                node.Features = lnwire.NewFeatureVector(
                        featureVector, featureNames,
                )

                node.Color, err = lncfg.ParseHexColor(rpcNode.Color)
                if err != nil {
                        return nil, err
                }

                if err := graphDB.AddNode(ctx, node); err != nil {
                        return nil, fmt.Errorf("unable to add node %v: %w",
                                rpcNode.PubKey, err)
                }

                log.Debugf("Imported node: %v", rpcNode.PubKey)
        }

        for _, rpcEdge := range graph.Edges {
                rpcEdge := rpcEdge

                edge := &models.ChannelEdgeInfo{
                        ChannelID: rpcEdge.ChannelId,
                        ChainHash: *s.cfg.ActiveNetParams.GenesisHash,
                        Capacity:  btcutil.Amount(rpcEdge.Capacity),
                }

                edge.NodeKey1Bytes, err = parsePubKey(rpcEdge.Node1Pub)
                if err != nil {
                        return nil, err
                }

                edge.NodeKey2Bytes, err = parsePubKey(rpcEdge.Node2Pub)
                if err != nil {
                        return nil, err
                }

                channelPoint, err := parseOutPoint(rpcEdge.ChanPoint)
                if err != nil {
                        return nil, err
                }
                edge.ChannelPoint = *channelPoint

                if err := graphDB.AddChannelEdge(ctx, edge); err != nil {
                        return nil, fmt.Errorf("unable to add edge %v: %w",
                                rpcEdge.ChanPoint, err)
                }

                makePolicy := func(rpcPolicy *lnrpc.RoutingPolicy) *models.ChannelEdgePolicy { //nolint:ll
                        policy := &models.ChannelEdgePolicy{
                                ChannelID: rpcEdge.ChannelId,
                                LastUpdate: time.Unix(
                                        int64(rpcPolicy.LastUpdate), 0,
                                ),
                                TimeLockDelta: uint16(
                                        rpcPolicy.TimeLockDelta,
                                ),
                                MinHTLC: lnwire.MilliSatoshi(
                                        rpcPolicy.MinHtlc,
                                ),
                                FeeBaseMSat: lnwire.MilliSatoshi(
                                        rpcPolicy.FeeBaseMsat,
                                ),
                                FeeProportionalMillionths: lnwire.MilliSatoshi(
                                        rpcPolicy.FeeRateMilliMsat,
                                ),
                        }
                        if rpcPolicy.MaxHtlcMsat > 0 {
                                policy.MaxHTLC = lnwire.MilliSatoshi(
                                        rpcPolicy.MaxHtlcMsat,
                                )
                                policy.MessageFlags |=
                                        lnwire.ChanUpdateRequiredMaxHtlc
                        }

                        return policy
                }

                if rpcEdge.Node1Policy != nil {
                        policy := makePolicy(rpcEdge.Node1Policy)
                        policy.ChannelFlags = 0
                        err := graphDB.UpdateEdgePolicy(ctx, policy)
                        if err != nil {
                                return nil, fmt.Errorf(
                                        "unable to update policy: %v", err)
                        }
                }

                if rpcEdge.Node2Policy != nil {
                        policy := makePolicy(rpcEdge.Node2Policy)
                        policy.ChannelFlags = 1
                        err := graphDB.UpdateEdgePolicy(ctx, policy)
                        if err != nil {
                                return nil, fmt.Errorf(
                                        "unable to update policy: %v", err)
                        }
                }

                log.Debugf("Added edge: %v", rpcEdge.ChannelId)
        }

        return &ImportGraphResponse{}, nil
}

// Quiesce initiates the quiescence process for the channel with the given
// channel ID. This method will block until the channel is fully quiesced.
func (s *Server) Quiesce(_ context.Context, in *QuiescenceRequest) (
        *QuiescenceResponse, error) {

        txid, err := lnrpc.GetChanPointFundingTxid(in.ChanId)
        if err != nil {
                return nil, err
        }

        op := wire.NewOutPoint(txid, in.ChanId.OutputIndex)
        cid := lnwire.NewChanIDFromOutPoint(*op)
        ln, err := s.cfg.Switch.GetLink(cid)
        if err != nil {
                return nil, err
        }

        select {
        case result := <-ln.InitStfu():
                mkResp := func(b lntypes.ChannelParty) *QuiescenceResponse {
                        return &QuiescenceResponse{Initiator: b.IsLocal()}
                }

                return fn.MapOk(mkResp)(result).Unpack()

        case <-s.quit:
                return nil, fmt.Errorf("server shutting down")
        }
}

// TriggerSweeper triggers the sweeper to immediately attempt to create and
// broadcast sweep transactions for all pending inputs. This is primarily used
// for testing to deterministically control when sweeps are broadcast.
//
// NOTE: Part of the DevServer interface.
func (s *Server) TriggerSweeper(_ context.Context,
        _ *TriggerSweeperRequest) (*TriggerSweeperResponse, error) {

        if s.cfg.Sweeper == nil {
                return nil, fmt.Errorf("sweeper not available")
        }

        numSweeps := s.cfg.Sweeper.TriggerSweep()

        log.Debugf("TriggerSweeper: triggered sweep of %d inputs", numSweeps)

        return &TriggerSweeperResponse{
                NumSweepsBroadcast: uint32(numSweeps),
        }, nil
}

1	//go:build dev
2	// +build dev
3
4	package devrpc
5
6	import (
7	"context"
8	"encoding/hex"
9	"fmt"
10	"strconv"
11	"strings"
12	"sync/atomic"
13	"time"
14
15	"github.com/btcsuite/btcd/btcutil"
16	"github.com/btcsuite/btcd/chaincfg/chainhash"
17	"github.com/btcsuite/btcd/wire"
18	"github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
19	"github.com/lightningnetwork/lnd/fn/v2"
20	"github.com/lightningnetwork/lnd/graph/db/models"
21	"github.com/lightningnetwork/lnd/lncfg"
22	"github.com/lightningnetwork/lnd/lnrpc"
23	"github.com/lightningnetwork/lnd/lntypes"
24	"github.com/lightningnetwork/lnd/lnwire"
25	"google.golang.org/grpc"
26	"gopkg.in/macaroon-bakery.v2/bakery"
27	)
28
29	const (
30	// subServerName is the name of the sub rpc server. We'll use this name
31	// to register ourselves, and we also require that the main
32	// SubServerConfigDispatcher instance recognize tt as the name of our
33	// RPC service.
34	subServerName = "DevRPC"
35	)
36
37	var (
38	// macPermissions maps RPC calls to the permissions they require.
39	macPermissions = map[string][]bakery.Op{
40	"/devrpc.Dev/ImportGraph": {{
41	Entity: "offchain",
42	Action: "write",
43	}},
44	"/devrpc.Dev/Quiesce": {{
45	Entity: "offchain",
46	Action: "write",
47	}},
48	"/devrpc.Dev/TriggerSweeper": {{
49	Entity: "onchain",
50	Action: "write",
51	}},
52	}
53	)
54
55	// ServerShell is a shell struct holding a reference to the actual sub-server.
56	// It is used to register the gRPC sub-server with the root server before we
57	// have the necessary dependencies to populate the actual sub-server.
58	type ServerShell struct {
59	DevServer
60	}
61
62	// Server is a sub-server of the main RPC server: the dev RPC. This sub
63	// RPC server allows developers to set and query LND state that is not possible
64	// during normal operation.
65	type Server struct {
66	started int32 // To be used atomically.
67	shutdown int32 // To be used atomically.
68	quit chan struct{}
69
70	// Required by the grpc-gateway/v2 library for forward compatibility.
71	// Must be after the atomically used variables to not break struct
72	// alignment.
73	UnimplementedDevServer
74
75	cfg *Config
76	}
77
78	// A compile time check to ensure that Server fully implements the
79	// DevServer gRPC service.
80	var _ DevServer = (*Server)(nil)
81
82	// New returns a new instance of the devrpc Dev sub-server. We also return the
83	// set of permissions for the macaroons that we may create within this method.
84	// If the macaroons we need aren't found in the filepath, then we'll create them
85	// on start up. If we're unable to locate, or create the macaroons we need, then
86	// we'll return with an error.
UNCOV 87	func New(cfg Config) (Server, lnrpc.MacaroonPerms, error) {	×
UNCOV 88	// We don't create any new macaroons for this subserver, instead reuse	×
UNCOV 89	// existing onchain/offchain permissions.	×
UNCOV 90	server := &Server{	×
UNCOV 91	quit: make(chan struct{}),	×
UNCOV 92	cfg: cfg,	×
UNCOV 93	}	×
UNCOV 94		×
UNCOV 95	return server, macPermissions, nil	×
UNCOV 96	}	×
97
98	// Start launches any helper goroutines required for the Server to function.
99	//
100	// NOTE: This is part of the lnrpc.SubServer interface.
UNCOV 101	func (s *Server) Start() error {	×
UNCOV 102	if atomic.AddInt32(&s.started, 1) != 1 {	×
103	return nil	×
104	}	×
105
UNCOV 106	return nil	×
107	}
108
109	// Stop signals any active goroutines for a graceful closure.
110	//
111	// NOTE: This is part of the lnrpc.SubServer interface.
UNCOV 112	func (s *Server) Stop() error {	×
UNCOV 113	if atomic.AddInt32(&s.shutdown, 1) != 1 {	×
114	return nil	×
115	}	×
116
UNCOV 117	close(s.quit)	×
UNCOV 118		×
UNCOV 119	return nil	×
120	}
121
122	// Name returns a unique string representation of the sub-server. This can be
123	// used to identify the sub-server and also de-duplicate them.
124	//
125	// NOTE: This is part of the lnrpc.SubServer interface.
UNCOV 126	func (s *Server) Name() string {	×
UNCOV 127	return subServerName	×
UNCOV 128	}	×
129
130	// RegisterWithRootServer will be called by the root gRPC server to direct a
131	// sub RPC server to register itself with the main gRPC root server. Until this
132	// is called, each sub-server won't be able to have
133	// requests routed towards it.
134	//
135	// NOTE: This is part of the lnrpc.GrpcHandler interface.
UNCOV 136	func (r ServerShell) RegisterWithRootServer(grpcServer grpc.Server) error {	×
UNCOV 137	// We make sure that we register it with the main gRPC server to ensure	×
UNCOV 138	// all our methods are routed properly.	×
UNCOV 139	RegisterDevServer(grpcServer, r)	×
UNCOV 140		×
UNCOV 141	log.Debugf("DEV RPC server successfully registered with root the " +	×
UNCOV 142	"gRPC server")	×
UNCOV 143		×
UNCOV 144	return nil	×
UNCOV 145	}	×
146
147	// RegisterWithRestServer will be called by the root REST mux to direct a sub
148	// RPC server to register itself with the main REST mux server. Until this is
149	// called, each sub-server won't be able to have requests routed towards it.
150	//
151	// NOTE: This is part of the lnrpc.GrpcHandler interface.
152	func (r *ServerShell) RegisterWithRestServer(ctx context.Context,
UNCOV 153	mux *runtime.ServeMux, dest string, opts []grpc.DialOption) error {	×
UNCOV 154		×
UNCOV 155	// We make sure that we register it with the main REST server to ensure	×
UNCOV 156	// all our methods are routed properly.	×
UNCOV 157	err := RegisterDevHandlerFromEndpoint(ctx, mux, dest, opts)	×
UNCOV 158	if err != nil {	×
159	log.Errorf("Could not register DEV REST server with the root "+	×
160	"REST server: %v", err)	×
161	return err	×
162	}	×
163
UNCOV 164	log.Debugf("DEV REST server successfully registered with the root " +	×
UNCOV 165	"REST server")	×
UNCOV 166	return nil	×
167	}
168
169	// CreateSubServer populates the subserver's dependencies using the passed
170	// SubServerConfigDispatcher. This method should fully initialize the
171	// sub-server instance, making it ready for action. It returns the macaroon
172	// permissions that the sub-server wishes to pass on to the root server for all
173	// methods routed towards it.
174	//
175	// NOTE: This is part of the lnrpc.GrpcHandler interface.
176	func (r *ServerShell) CreateSubServer(configRegistry lnrpc.SubServerConfigDispatcher) (
UNCOV 177	lnrpc.SubServer, lnrpc.MacaroonPerms, error) {	×
UNCOV 178		×
UNCOV 179	subServer, macPermissions, err := createNewSubServer(configRegistry)	×
UNCOV 180	if err != nil {	×
181	return nil, nil, err	×
182	}	×
183
UNCOV 184	r.DevServer = subServer	×
UNCOV 185	return subServer, macPermissions, nil	×
186	}
187
188	func parseOutPoint(s string) (*wire.OutPoint, error) {	×
189	split := strings.Split(s, ":")	×
190	if len(split) != 2 {	×
191	return nil, fmt.Errorf("expecting outpoint to be in format of: " +	×
192	"txid:index")	×
193	}	×
194
195	index, err := strconv.ParseInt(split[1], 10, 32)	×
196	if err != nil {	×
197	return nil, fmt.Errorf("unable to decode output index: %w", err)	×
198	}	×
199
200	txid, err := chainhash.NewHashFromStr(split[0])	×
201	if err != nil {	×
202	return nil, fmt.Errorf("unable to parse hex string: %w", err)	×
203	}	×
204
205	return &wire.OutPoint{	×
206	Hash: *txid,	×
207	Index: uint32(index),	×
208	}, nil	×
209	}
210
211	func parsePubKey(pubKeyStr string) ([33]byte, error) {	×
212	var pubKey [33]byte	×
213	pubKeyBytes, err := hex.DecodeString(pubKeyStr)	×
214	if err != nil \|\| len(pubKeyBytes) != 33 {	×
215	return pubKey, fmt.Errorf("invalid pubkey: %v", pubKeyStr)	×
216	}	×
217
218	copy(pubKey[:], pubKeyBytes)	×
219	return pubKey, nil	×
220	}
221
222	// ImportGraph imports a graph dump (without auth proofs).
223	//
224	// NOTE: Part of the DevServer interface.
225	func (s *Server) ImportGraph(ctx context.Context,
226	graph lnrpc.ChannelGraph) (ImportGraphResponse, error) {	×
227		×
228	// Obtain the pointer to the global singleton channel graph.	×
229	graphDB := s.cfg.GraphDB	×
230		×
231	var err error	×
232	for _, rpcNode := range graph.Nodes {	×
233	node := &models.Node{	×
234	HaveNodeAnnouncement: true,	×
235	LastUpdate: time.Unix(	×
236	int64(rpcNode.LastUpdate), 0,	×
237	),	×
238	Alias: rpcNode.Alias,	×
239	}	×
240		×
241	node.PubKeyBytes, err = parsePubKey(rpcNode.PubKey)	×
242	if err != nil {	×
243	return nil, err	×
244	}	×
245
246	featureBits := make([]lnwire.FeatureBit, 0, len(rpcNode.Features))	×
247	featureNames := make(map[lnwire.FeatureBit]string)	×
248		×
249	for featureBit, feature := range rpcNode.Features {	×
250	featureBits = append(	×
251	featureBits, lnwire.FeatureBit(featureBit),	×
252	)	×
253		×
254	featureNames[lnwire.FeatureBit(featureBit)] = feature.Name	×
255	}	×
256
257	featureVector := lnwire.NewRawFeatureVector(featureBits...)	×
258	node.Features = lnwire.NewFeatureVector(	×
259	featureVector, featureNames,	×
260	)	×
261		×
262	node.Color, err = lncfg.ParseHexColor(rpcNode.Color)	×
263	if err != nil {	×
264	return nil, err	×
265	}	×
266
267	if err := graphDB.AddNode(ctx, node); err != nil {	×
268	return nil, fmt.Errorf("unable to add node %v: %w",	×
269	rpcNode.PubKey, err)	×
270	}	×
271
272	log.Debugf("Imported node: %v", rpcNode.PubKey)	×
273	}
274
275	for _, rpcEdge := range graph.Edges {	×
276	rpcEdge := rpcEdge	×
277		×
278	edge := &models.ChannelEdgeInfo{	×
279	ChannelID: rpcEdge.ChannelId,	×
280	ChainHash: *s.cfg.ActiveNetParams.GenesisHash,	×
281	Capacity: btcutil.Amount(rpcEdge.Capacity),	×
282	}	×
283		×
284	edge.NodeKey1Bytes, err = parsePubKey(rpcEdge.Node1Pub)	×
285	if err != nil {	×
286	return nil, err	×
287	}	×
288
289	edge.NodeKey2Bytes, err = parsePubKey(rpcEdge.Node2Pub)	×
290	if err != nil {	×
291	return nil, err	×
292	}	×
293
294	channelPoint, err := parseOutPoint(rpcEdge.ChanPoint)	×
295	if err != nil {	×
296	return nil, err	×
297	}	×
298	edge.ChannelPoint = *channelPoint	×
299		×
300	if err := graphDB.AddChannelEdge(ctx, edge); err != nil {	×
301	return nil, fmt.Errorf("unable to add edge %v: %w",	×
302	rpcEdge.ChanPoint, err)	×
303	}	×
304
305	makePolicy := func(rpcPolicy lnrpc.RoutingPolicy) models.ChannelEdgePolicy { //nolint:ll	×
306	policy := &models.ChannelEdgePolicy{	×
307	ChannelID: rpcEdge.ChannelId,	×
308	LastUpdate: time.Unix(	×
309	int64(rpcPolicy.LastUpdate), 0,	×
310	),	×
311	TimeLockDelta: uint16(	×
312	rpcPolicy.TimeLockDelta,	×
313	),	×
314	MinHTLC: lnwire.MilliSatoshi(	×
315	rpcPolicy.MinHtlc,	×
316	),	×
317	FeeBaseMSat: lnwire.MilliSatoshi(	×
318	rpcPolicy.FeeBaseMsat,	×
319	),	×
320	FeeProportionalMillionths: lnwire.MilliSatoshi(	×
321	rpcPolicy.FeeRateMilliMsat,	×
322	),	×
323	}	×
324	if rpcPolicy.MaxHtlcMsat > 0 {	×
325	policy.MaxHTLC = lnwire.MilliSatoshi(	×
326	rpcPolicy.MaxHtlcMsat,	×
327	)	×
328	policy.MessageFlags \|=	×
329	lnwire.ChanUpdateRequiredMaxHtlc	×
330	}	×
331
332	return policy	×
333	}
334
335	if rpcEdge.Node1Policy != nil {	×
336	policy := makePolicy(rpcEdge.Node1Policy)	×
337	policy.ChannelFlags = 0	×
338	err := graphDB.UpdateEdgePolicy(ctx, policy)	×
339	if err != nil {	×
340	return nil, fmt.Errorf(	×
341	"unable to update policy: %v", err)	×
342	}	×
343	}
344
345	if rpcEdge.Node2Policy != nil {	×
346	policy := makePolicy(rpcEdge.Node2Policy)	×
347	policy.ChannelFlags = 1	×
348	err := graphDB.UpdateEdgePolicy(ctx, policy)	×
349	if err != nil {	×
350	return nil, fmt.Errorf(	×
351	"unable to update policy: %v", err)	×
352	}	×
353	}
354
355	log.Debugf("Added edge: %v", rpcEdge.ChannelId)	×
356	}
357
358	return &ImportGraphResponse{}, nil	×
359	}
360
361	// Quiesce initiates the quiescence process for the channel with the given
362	// channel ID. This method will block until the channel is fully quiesced.
363	func (s Server) Quiesce(_ context.Context, in QuiescenceRequest) (
UNCOV 364	*QuiescenceResponse, error) {	×
UNCOV 365		×
UNCOV 366	txid, err := lnrpc.GetChanPointFundingTxid(in.ChanId)	×
UNCOV 367	if err != nil {	×
368	return nil, err	×
369	}	×
370
UNCOV 371	op := wire.NewOutPoint(txid, in.ChanId.OutputIndex)	×
UNCOV 372	cid := lnwire.NewChanIDFromOutPoint(*op)	×
UNCOV 373	ln, err := s.cfg.Switch.GetLink(cid)	×
UNCOV 374	if err != nil {	×
375	return nil, err	×
376	}	×
377
UNCOV 378	select {	×
UNCOV 379	case result := <-ln.InitStfu():	×
UNCOV 380	mkResp := func(b lntypes.ChannelParty) *QuiescenceResponse {	×
UNCOV 381	return &QuiescenceResponse{Initiator: b.IsLocal()}	×
UNCOV 382	}	×
383
UNCOV 384	return fn.MapOk(mkResp)(result).Unpack()	×
385
386	case <-s.quit:	×
387	return nil, fmt.Errorf("server shutting down")	×
388	}
389	}
390
391	// TriggerSweeper triggers the sweeper to immediately attempt to create and
392	// broadcast sweep transactions for all pending inputs. This is primarily used
393	// for testing to deterministically control when sweeps are broadcast.
394	//
395	// NOTE: Part of the DevServer interface.
396	func (s *Server) TriggerSweeper(_ context.Context,
NEW 397	_ TriggerSweeperRequest) (TriggerSweeperResponse, error) {	×
NEW 398		×
NEW 399	if s.cfg.Sweeper == nil {	×
NEW 400	return nil, fmt.Errorf("sweeper not available")	×
NEW 401	}	×
402
NEW 403	numSweeps := s.cfg.Sweeper.TriggerSweep()	×
NEW 404		×
NEW 405	log.Debugf("TriggerSweeper: triggered sweep of %d inputs", numSweeps)	×
NEW 406		×
NEW 407	return &TriggerSweeperResponse{	×
NEW 408	NumSweepsBroadcast: uint32(numSweeps),	×
NEW 409	}, nil	×
410	}

lightningnetwork / lnd / 18852986778

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